gavindiaz/mt5cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: consolidate duplicate test cases into parameterized tests (#91)

Browse Source 
* test: consolidate duplicate test cases into parameterized tests

- Combine test_symbol_info, test_symbol_info_tick, test_market_book into test_symbol_command with parametrize
- Combine test_order_check variants into test_order_request with parametrize
- Combine test_order_send variants into test_order_request with parametrize
- Consolidate grafana view skip tests into test_grafana_view_skipped_when_required_cols_missing
- Consolidate grafana index skip tests into test_index_skipped_when_cols_missing
- Consolidate _NoOp method tests into test_method_is_noop
- Consolidate record_*_update noop tests into test_record_update_noop_before_configure

This improves test maintainability by reducing duplication while maintaining coverage.

* test: parametrize remaining duplicated cases

* test: parameterize remaining duplicate test cases

* test: parameterize remaining duplicate test cases

* test: parameterize remaining duplicate test cases

* test: parameterize remaining duplicate test cases

* test: parameterize count and start_pos validation tests

* test: parameterize remaining duplicate test cases

- tests/test_utils.py: merge valid string/integer cases for
  parse_timeframe and parse_tick_flags.
- tests/test_cli.py: consolidate collect-history --if-exists append/fail
  tests and drop the duplicate collect-history --dataset ticks default
  flags=-1 assertion.
- tests/test_trading.py: parametrize buy/sell projected margin ratio,
  invalid place_market_order mode validation, core
  calculate_volume_by_margin boundary cases, and zero-ratio
  calculate_margin_and_volume sizing cases.

* test: parameterize remaining duplicate test cases

Consolidate the last high-value duplicate test cases outside the
already-changed areas while keeping semantics and 100% coverage.

- tests/test_sdk.py:TestMt5CliClient - merge copy_rates_range,
  copy_ticks_from, history_orders, and latest_rates delegation/
  normalization tests into test_method_delegates_with_normalization
  parameterized by call/expected_method/expected_kwargs with readable
  ids for each normalization path.
- tests/test_sdk.py:TestSubstituteEnvPlaceholders - merge
  brace/whole-dollar/plain/partial substitution cases into
  test_substitute_env_placeholders, and missing-env cases into
  test_substitute_env_placeholders_raises_on_missing_env.
- tests/test_trading.py:TestVolumeAndExecution - merge buy/sell
  trailing-stop main cases into test_calculate_trailing_stop_updates_by_side,
  opposite-side invalid-quote cases into
  test_calculate_trailing_stop_updates_ignores_opposite_side_price,
  and the two mixed-positions scenarios into the parametrized
  test_calculate_trailing_stop_updates_mixed_positions_skip_invalid_side.
  All ids explicitly describe the side and behavior.

* test: parameterize collect-history and telemetry update tests

Consolidate duplicate default-vs-ticks collect-history cases in CLI and
SDK tests, and merge history/snapshot record_*_update success and
failure tests in test_telemetry.py.

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: parameterize telemetry record_*_state gauge tests

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: parameterize Grafana example file checks

* test: centralize remaining parametrized cases

* test: add parametrized replacements for remaining cases

* test: fix remaining parametrized lint

* Fix Ruff warnings

* test: remove duplicate parametrization sweep file

* test: remove duplicate-test deselection hook

* test: restore conftest formatting

* test: add remaining parametrized regression coverage

* test: fix lint issues in parametrization completion tests

* test: avoid dynamic SQL in parametrization completion tests

* test: format parametrization completion tests

* test: fix lint and pyright issues in parametrization coverage

* Fix a Pyright error

* Update .agents/skills/pr-feedback-triage/SKILL.md

* test: dissolve parametrize-completion catch-all into owning test files

Move the remaining consolidated cases from test_parametrize_completion.py
into their owning modules and delete the file:

- test_cli.py: merge snapshot/grafana-schema --publish-copy gating into
  one parametrized test.
- test_trading.py: merge calculate_spread_ratio numeric/numeric-string
  cases; fold non-finite volume_max into
  test_normalize_order_volume_deterministic; merge estimate_order_margin
  invalid-volume cases (0.0/nan/inf); merge calculate_positions_margin
  invalid-row filtering cases.
- test_history.py: parametrize resolve_history_tick_flags and
  resolve_granularity_name.
- test_sdk.py: parametrize collect_latest_closed_rates_for_accounts
  empty-effective-frame cases.
- test_grafana.py: merge account/terminal snapshot insert tests inside
  TestSnapshotInserts.

All remaining cases in the removed file were exact duplicates of
existing coverage, so no new test scenarios are introduced. 100%
coverage and all checks (ruff, pyright, pytest, local-qa) pass.

* test: parameterize remaining account-event, snapshot, and CLI edge cases

Consolidates six more pairs of duplicate tests flagged during PR #91
review into pytest.mark.parametrize tables (incremental history_deals
account-event edge cases, dedup scope column variants, snapshot view
status filtering, snapshot kwargs forwarding, injected-client
lifecycle, and close-positions --yes gate), preserving explicit ids
and 100% coverage.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* test: consolidate build_config login coercion into parametrized cases

Fold separate backward-compat and env-expansion tests into named
pytest.param rows for clearer failure output.

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: consolidate deduplication and publish test cases into parametrized variants

Merge related test cases into single parametrized tests:
- Consolidate collect-history validation tests into a single parametrized case
- Merge snapshot-view skip conditions into one test with parametrized inputs
- Combine publish_grafana_copy target variations into one parametrized test
- Consolidate incremental-start and drop-duplicates error cases
- Remove duplicate test_drop_duplicates_rejects_invalid_identifiers

Reduces test file duplication while maintaining full coverage.

* test: consolidate remaining review-identified parameterized duplicates

Merge close-position filter, margin-ratio suppress/reraise, SQLite path
validation, and grafana_symbol_pnl schema cases into clearer parametrized tests.

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: restore default if_exists coverage in SQLite append test

The default-append parametrized case now omits if_exists so the API default
remains exercised instead of passing IfExists.APPEND explicitly.

Co-authored-by: Cursor <cursoragent@cursor.com>

* test: tighten trading parametrization cleanup

* test: consolidate remaining parameterized cases

* test: consolidate additional duplicate cases into parametrized tests

Merge overlapping success-path tests in contracts, history, SDK, and
trading modules while preserving 100% coverage and explicit test ids.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: agent <agent@localhost>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Daichi Narushima
2026-07-03 03:54:26 +09:00
committed by GitHub
parent a05b6b896d
commit a5c2aa8e8f
10 changed files with 2949 additions and 3136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_cli.py
+477 -786
View File
File diff suppressed because it is too large Load Diff
tests/test_contracts.py
+26 -18
View File
@@ -224,10 +224,18 @@ def test_parse_date_range_rejects_inverted_bounds() -> None:
parse_date_range("2024-02-01", "2024-01-01")
def test_recent_window_builds_trailing_bounds() -> None:
"""Recent windows end at the provided timestamp."""
@pytest.mark.parametrize(
"kwargs",
[
{"hours": 24},
{"seconds": 3600},
],
ids=["hours", "seconds"],
)
def test_recent_window_success_cases(kwargs: dict[str, int]) -> None:
"""Recent windows end at the provided timestamp for both duration inputs."""
end = datetime(2024, 1, 2, tzinfo=UTC)
start, resolved_end = recent_window(hours=24, date_to=end)
start, resolved_end = recent_window(date_to=end, **kwargs)
assert resolved_end == end
assert start < end
@@ -339,22 +347,22 @@ def test_ensure_utc_handles_naive_and_aware_datetimes() -> None:
assert ensure_utc("2024-01-01T00:00:00+00:00").tzinfo == UTC
def test_recent_window_validation_errors() -> None:
@pytest.mark.parametrize(
("kwargs", "match"),
[
({}, "exactly one"),
({"hours": 1, "seconds": 1}, "exactly one"),
({"hours": 0}, "positive"),
],
ids=["no-duration", "both-hours-and-seconds", "non-positive-duration"],
)
def test_recent_window_validation_errors(
kwargs: dict[str, object],
match: str,
) -> None:
"""Recent window helpers validate mutually exclusive length arguments."""
with pytest.raises(ValueError, match="exactly one"):
recent_window()
with pytest.raises(ValueError, match="exactly one"):
recent_window(hours=1, seconds=1)
with pytest.raises(ValueError, match="positive"):
recent_window(hours=0)
def test_recent_window_supports_seconds_argument() -> None:
"""Recent windows can be built from a seconds-based length."""
end = datetime(2024, 1, 2, tzinfo=UTC)
start, resolved_end = recent_window(seconds=3600, date_to=end)
assert resolved_end == end
assert start < end
with pytest.raises(ValueError, match=match):
recent_window(**kwargs) # type: ignore[arg-type]
def test_parse_date_range_returns_ordered_bounds() -> None:
tests/test_examples.py
+67 -50
View File
@@ -4,72 +4,89 @@ from __future__ import annotations
import json
from pathlib import Path
from typing import TYPE_CHECKING
import pytest
if TYPE_CHECKING:
from collections.abc import Iterator
_EXAMPLES_DIR = Path(__file__).parent.parent / "examples" / "grafana"
_DASHBOARDS_DIR = _EXAMPLES_DIR / "dashboards"
def _dashboard_json_files() -> list[Path]:
"""Return bundled Grafana dashboard JSON files in deterministic order."""
return sorted(_DASHBOARDS_DIR.glob("*.json"))
@pytest.fixture(params=_dashboard_json_files(), ids=lambda path: path.name)
def dashboard_path(request: pytest.FixtureRequest) -> Iterator[Path]:
"""Yield one bundled Grafana dashboard JSON file per test case."""
return request.param
class TestGrafanaExamples:
"""Validate structure and content of bundled Grafana example files."""
def test_dashboard_json_files_are_valid_json(self) -> None:
"""All dashboard JSON files parse without error."""
paths = list(_DASHBOARDS_DIR.glob("*.json"))
assert paths, "No dashboard JSON files found"
for path in paths:
content = path.read_text(encoding="utf-8")
obj = json.loads(content)
assert isinstance(obj, dict), f"{path.name} root must be a JSON object"
def test_dashboard_json_files_are_present(self) -> None:
"""At least one dashboard JSON file is bundled."""
assert _dashboard_json_files(), "No dashboard JSON files found"
def test_dashboard_json_has_no_private_placeholders(self) -> None:
def test_dashboard_json_file_is_valid_json(self, dashboard_path: Path) -> None:
"""Each dashboard JSON file parses to a JSON object."""
content = dashboard_path.read_text(encoding="utf-8")
obj = json.loads(content)
assert isinstance(obj, dict), (
f"{dashboard_path.name} root must be a JSON object"
)
@pytest.mark.parametrize("private_pattern", ["password", "api_key", "apikey"])
def test_dashboard_json_has_no_private_placeholders(
self,
dashboard_path: Path,
private_pattern: str,
) -> None:
"""Dashboard JSON files contain no obvious credential placeholders."""
private_patterns = ["password", "api_key", "apikey"]
for path in _DASHBOARDS_DIR.glob("*.json"):
content = path.read_text(encoding="utf-8").lower()
for pat in private_patterns:
assert pat not in content, f"{path.name} contains {pat!r}"
content = dashboard_path.read_text(encoding="utf-8").lower()
assert private_pattern not in content, (
f"{dashboard_path.name} contains {private_pattern!r}"
)
def test_dashboard_json_uses_grafana_views(self) -> None:
"""All dashboard JSON files query grafana_* views."""
for path in _DASHBOARDS_DIR.glob("*.json"):
content = path.read_text(encoding="utf-8")
assert "grafana_" in content, (
f"{path.name} must contain queries against grafana_* views"
)
def test_dashboard_json_uses_grafana_views(self, dashboard_path: Path) -> None:
"""Each dashboard JSON file queries grafana_* views."""
content = dashboard_path.read_text(encoding="utf-8")
assert "grafana_" in content, (
f"{dashboard_path.name} must contain queries against grafana_* views"
)
def test_dashboard_json_has_uid(self) -> None:
"""All dashboard JSON files have a non-empty uid field."""
for path in _DASHBOARDS_DIR.glob("*.json"):
obj = json.loads(path.read_text(encoding="utf-8"))
assert obj.get("uid"), f"{path.name} must have a uid"
def test_dashboard_json_has_title(self) -> None:
"""All dashboard JSON files have a non-empty title field."""
for path in _DASHBOARDS_DIR.glob("*.json"):
obj = json.loads(path.read_text(encoding="utf-8"))
assert obj.get("title"), f"{path.name} must have a title"
@pytest.mark.parametrize("field", ["uid", "title"])
def test_dashboard_json_has_required_field(
self,
dashboard_path: Path,
field: str,
) -> None:
"""Each dashboard JSON file has a non-empty uid and title field."""
obj = json.loads(dashboard_path.read_text(encoding="utf-8"))
assert obj.get(field), f"{dashboard_path.name} must have a {field}"
def test_expected_dashboards_present(self) -> None:
"""The three expected dashboard files are present."""
names = {p.name for p in _DASHBOARDS_DIR.glob("*.json")}
names = {p.name for p in _dashboard_json_files()}
assert "mt5cli-overview.json" in names
assert "mt5cli-trades.json" in names
assert "mt5cli-market.json" in names
def test_readme_exists(self) -> None:
"""examples/grafana/README.md is present."""
assert (_EXAMPLES_DIR / "README.md").is_file()
def test_compose_file_exists(self) -> None:
"""examples/grafana/compose.yml is present."""
assert (_EXAMPLES_DIR / "compose.yml").is_file()
def test_datasource_provisioning_exists(self) -> None:
"""Datasource provisioning YAML is present."""
assert (
_EXAMPLES_DIR / "provisioning" / "datasources" / "mt5cli-sqlite.yml"
).is_file()
def test_dashboard_provisioning_exists(self) -> None:
"""Dashboard provisioning YAML is present."""
assert (_EXAMPLES_DIR / "provisioning" / "dashboards" / "mt5cli.yml").is_file()
@pytest.mark.parametrize(
"relative_path",
[
"README.md",
"compose.yml",
"provisioning/datasources/mt5cli-sqlite.yml",
"provisioning/dashboards/mt5cli.yml",
],
ids=["readme", "compose", "datasource-provisioning", "dashboard-provisioning"],
)
def test_expected_file_exists(self, relative_path: str) -> None:
"""Bundled Grafana example support files are present."""
assert (_EXAMPLES_DIR / relative_path).is_file()
tests/test_grafana.py
+371 -419
View File
@@ -12,7 +12,7 @@ import pandas as pd
import pytest
if TYPE_CHECKING:
from collections.abc import Iterator
from collections.abc import Callable, Iterator
from mt5cli.grafana import (
_build_snapshot_view, # type: ignore[reportPrivateUsage]
@@ -75,6 +75,14 @@ def _make_history_deals_minimal(conn: sqlite3.Connection) -> None:
)
def _make_history_deals_symbol_pnl_minimal(conn: sqlite3.Connection) -> None:
"""history_deals with entry but without volume and price columns."""
conn.execute(
"CREATE TABLE history_deals"
" (time TEXT, symbol TEXT, profit REAL, type INTEGER, entry INTEGER)"
)
def _make_history_orders_table(conn: sqlite3.Connection) -> None:
conn.execute(
"CREATE TABLE history_orders"
@@ -82,6 +90,9 @@ def _make_history_orders_table(conn: sqlite3.Connection) -> None:
)
_TIMESTAMP_TIME_SETUP: pd.Timestamp = pd.Timestamp("2024-01-15 10:30:00", tz="UTC")
# ---------------------------------------------------------------------------
# TestSnapshotTables
# ---------------------------------------------------------------------------
@@ -204,169 +215,97 @@ class TestGrafanaViews:
create_grafana_views(conn)
assert "grafana_rates" not in _get_names(conn, "view")
def test_grafana_rates_skipped_when_required_cols_missing(
@pytest.mark.parametrize(
("ddl", "view_name"),
[
("CREATE TABLE rates (open REAL)", "grafana_rates"),
("CREATE TABLE ticks (bid REAL)", "grafana_ticks"),
("CREATE TABLE history_deals (symbol TEXT)", "grafana_history_deals"),
("CREATE TABLE history_orders (symbol TEXT)", "grafana_history_orders"),
("CREATE TABLE history_deals (symbol TEXT)", "grafana_trade_deals"),
("CREATE TABLE history_deals (symbol TEXT)", "grafana_cash_events"),
(
"CREATE TABLE history_deals (time TEXT, type INTEGER)",
"grafana_realized_pnl",
),
(
(
"CREATE TABLE history_deals"
" (time TEXT, symbol TEXT, profit REAL, type INTEGER)"
),
"grafana_realized_pnl",
),
(
"CREATE TABLE history_deals (time TEXT, type INTEGER)",
"grafana_symbol_pnl",
),
("CREATE TABLE history_deals (time TEXT)", "grafana_trade_stats"),
],
ids=[
"rates-cols",
"ticks-cols",
"history_deals-time",
"history_orders-time_setup",
"trade_deals-cols",
"cash_events-cols",
"realized_pnl-cols",
"realized_pnl-entry",
"symbol_pnl-cols",
"trade_stats-cols",
],
)
def test_grafana_view_skipped_when_required_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
ddl: str,
view_name: str,
) -> None:
"""grafana_rates is skipped when rates table lacks required columns."""
conn.execute("CREATE TABLE rates (open REAL)")
"""A grafana view is skipped (with warning) when source columns are missing."""
conn.execute(ddl)
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_rates" not in _get_names(conn, "view")
assert "Skipping grafana_rates" in caplog.text
assert view_name not in _get_names(conn, "view")
assert f"Skipping {view_name}" in caplog.text
def test_grafana_ticks_skipped_when_cols_missing(
@pytest.mark.parametrize(
("setup_deals", "optional_cols"),
[
pytest.param(
_make_history_deals_symbol_pnl_minimal, set[str](), id="minimal"
),
pytest.param(
_make_history_deals_full,
{"volume", "price"},
id="full",
),
],
)
def test_grafana_symbol_pnl_schema(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
setup_deals: Callable[[sqlite3.Connection], None],
optional_cols: set[str],
) -> None:
"""grafana_ticks is skipped when ticks table lacks required columns."""
conn.execute("CREATE TABLE ticks (bid REAL)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_ticks" not in _get_names(conn, "view")
assert "Skipping grafana_ticks" in caplog.text
def test_grafana_history_deals_skipped_when_time_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_history_deals is skipped when history_deals.time is missing."""
conn.execute("CREATE TABLE history_deals (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_history_deals" not in _get_names(conn, "view")
assert "Skipping grafana_history_deals" in caplog.text
def test_grafana_history_orders_skipped_when_time_setup_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_history_orders is skipped when time_setup is absent."""
conn.execute("CREATE TABLE history_orders (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_history_orders" not in _get_names(conn, "view")
assert "Skipping grafana_history_orders" in caplog.text
def test_grafana_trade_deals_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_trade_deals is skipped when history_deals missing time/type."""
conn.execute("CREATE TABLE history_deals (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_trade_deals" not in _get_names(conn, "view")
def test_grafana_cash_events_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_cash_events is skipped when history_deals missing time/type."""
conn.execute("CREATE TABLE history_deals (symbol TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_cash_events" not in _get_names(conn, "view")
def test_grafana_realized_pnl_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_realized_pnl is skipped when history_deals missing required cols."""
conn.execute("CREATE TABLE history_deals (time TEXT, type INTEGER)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_realized_pnl" not in _get_names(conn, "view")
assert "Skipping grafana_realized_pnl" in caplog.text
def test_grafana_realized_pnl_skipped_when_entry_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_realized_pnl is skipped when entry column is absent."""
_make_history_deals_minimal(conn)
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_realized_pnl" not in _get_names(conn, "view")
assert "Skipping grafana_realized_pnl" in caplog.text
def test_grafana_symbol_pnl_skipped_when_required_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_symbol_pnl is skipped when required columns are absent."""
conn.execute("CREATE TABLE history_deals (time TEXT, type INTEGER)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_symbol_pnl" not in _get_names(conn, "view")
assert "Skipping grafana_symbol_pnl" in caplog.text
def test_grafana_symbol_pnl_without_volume_and_price(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_symbol_pnl is created with only required columns."""
conn.execute(
"CREATE TABLE history_deals"
" (time TEXT, symbol TEXT, profit REAL, type INTEGER, entry INTEGER)"
)
"""grafana_symbol_pnl is created and includes optional columns when present."""
setup_deals(conn)
create_grafana_views(conn)
assert "grafana_symbol_pnl" in _get_names(conn, "view")
def test_grafana_symbol_pnl_with_volume_and_price(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_symbol_pnl includes volume and price columns when present."""
_make_history_deals_full(conn)
create_grafana_views(conn)
assert "grafana_symbol_pnl" in _get_names(conn, "view")
# View columns include volume and price
cols = {row[1] for row in conn.execute("PRAGMA table_info(grafana_symbol_pnl)")}
assert "volume" in cols
assert "price" in cols
assert optional_cols.issubset(cols)
def test_grafana_trade_stats_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
) -> None:
"""grafana_trade_stats is skipped when history_deals missing required cols."""
conn.execute("CREATE TABLE history_deals (time TEXT)")
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
create_grafana_views(conn)
assert "grafana_trade_stats" not in _get_names(conn, "view")
assert "Skipping grafana_trade_stats" in caplog.text
def test_grafana_trade_stats_without_entry_col(
@pytest.mark.parametrize(
"setup_deals",
[_make_history_deals_minimal, _make_history_deals_full],
ids=["minimal", "full"],
)
def test_grafana_trade_stats_static_summary(
self,
conn: sqlite3.Connection,
setup_deals: Callable[[sqlite3.Connection], None],
) -> None:
"""grafana_trade_stats is a static summary view with no time column."""
_make_history_deals_minimal(conn)
create_grafana_views(conn)
assert "grafana_trade_stats" in _get_names(conn, "view")
cols = {
row[1] for row in conn.execute("PRAGMA table_info(grafana_trade_stats)")
}
assert "time" not in cols
assert "symbol" in cols
def test_grafana_trade_stats_with_entry_col(
self,
conn: sqlite3.Connection,
) -> None:
"""grafana_trade_stats is a static summary view with no time column."""
_make_history_deals_full(conn)
setup_deals(conn)
create_grafana_views(conn)
assert "grafana_trade_stats" in _get_names(conn, "view")
cols = {
@@ -405,116 +344,126 @@ class TestGrafanaViews:
assert "time" in cols
assert "run_id" in cols
def test_build_snapshot_view_skips_when_snapshot_runs_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""_build_snapshot_view skips view when snapshot_runs has wrong columns."""
conn.execute("CREATE TABLE only_run (run_id INTEGER NOT NULL)")
conn.execute("CREATE TABLE snapshot_runs (foo TEXT)")
_build_snapshot_view(conn, "test_view", "only_run")
views = _get_names(conn, "view")
assert "test_view" not in views
def test_build_snapshot_view_skips_when_run_id_col_missing(
@pytest.mark.parametrize(
("use_snapshot_tables", "table_ddl", "table_name", "log_fragment"),
[
pytest.param(
False,
"CREATE TABLE only_run (run_id INTEGER NOT NULL)",
"only_run",
"snapshot_runs missing required columns",
id="snapshot-runs-wrong-columns",
),
pytest.param(
True,
"CREATE TABLE no_run_id (symbol TEXT)",
"no_run_id",
"missing run_id column",
id="table-missing-run-id",
),
],
)
def test_build_snapshot_view_skips_negative_cases(
self,
conn: sqlite3.Connection,
caplog: pytest.LogCaptureFixture,
use_snapshot_tables: bool,
table_ddl: str,
table_name: str,
log_fragment: str,
) -> None:
"""_build_snapshot_view skips view when the table lacks run_id."""
create_snapshot_tables(conn)
conn.execute("CREATE TABLE no_run_id (symbol TEXT)")
"""_build_snapshot_view skips view creation for invalid table/run metadata."""
if use_snapshot_tables:
create_snapshot_tables(conn)
else:
conn.execute("CREATE TABLE snapshot_runs (foo TEXT)")
conn.execute(table_ddl)
with caplog.at_level(logging.WARNING, logger="mt5cli.grafana"):
_build_snapshot_view(conn, "test_view", "no_run_id")
_build_snapshot_view(conn, "test_view", table_name)
assert "test_view" not in _get_names(conn, "view")
assert "missing run_id column" in caplog.text
assert log_fragment in caplog.text
def test_snapshot_view_excludes_failed_run_rows(
@pytest.mark.parametrize(
("started_at", "status", "message", "keep_row"),
[
pytest.param(
1000,
"error",
"terminal offline",
False,
id="excludes-failed-run-rows",
),
pytest.param(2000, "ok", None, True, id="includes-ok-run-rows"),
],
)
def test_snapshot_view_filters_rows_by_run_status(
self,
conn: sqlite3.Connection,
started_at: int,
status: str,
message: str | None,
keep_row: bool,
) -> None:
"""Snapshot views hide rows from failed runs."""
"""Snapshot views expose rows only from successful runs."""
create_snapshot_tables(conn)
run_id = start_snapshot_run(conn, 1000)
run_id = start_snapshot_run(conn, started_at)
conn.execute(
"INSERT INTO account_snapshots"
" (run_id, login, balance, equity, margin, margin_free, profit)"
" VALUES (?, 12345, 10000.0, 9800.0, 200.0, 9600.0, -200.0)",
(run_id,),
)
record_snapshot_run(conn, run_id, "error", "terminal offline")
create_grafana_views(conn)
rows = conn.execute("SELECT * FROM grafana_account_snapshots").fetchall()
assert rows == []
def test_snapshot_view_includes_ok_run_rows(
self,
conn: sqlite3.Connection,
) -> None:
"""Snapshot views show rows from successful runs and expose run_id."""
create_snapshot_tables(conn)
run_id = start_snapshot_run(conn, 2000)
conn.execute(
"INSERT INTO account_snapshots"
" (run_id, login, balance, equity, margin, margin_free, profit)"
" VALUES (?, 12345, 10000.0, 9800.0, 200.0, 9600.0, -200.0)",
(run_id,),
)
record_snapshot_run(conn, run_id, "ok")
record_snapshot_run(conn, run_id, status, message)
create_grafana_views(conn)
rows = conn.execute(
"SELECT time, run_id, login FROM grafana_account_snapshots"
).fetchall()
assert rows == [(2000, run_id, 12345)]
cols = {
row[1]
for row in conn.execute("PRAGMA table_info(grafana_account_snapshots)")
}
assert "run_id" in cols
if keep_row:
assert rows == [(started_at, run_id, 12345)]
cols = {
row[1]
for row in conn.execute("PRAGMA table_info(grafana_account_snapshots)")
}
assert "run_id" in cols
else:
assert rows == []
def test_snapshot_view_same_second_ok_and_error_no_cross_contamination(
@pytest.mark.parametrize(
("observed_at", "runs", "expected_rows"),
[
pytest.param(
3000,
[("error", 99), ("ok", 12345)],
[(12345,)],
id="same-second-error-and-ok-exposes-only-ok-row",
),
pytest.param(
4000,
[("ok", 1), ("ok", 2)],
[(1,), (2,)],
id="same-second-two-ok-runs-no-duplication",
),
],
)
def test_snapshot_view_same_second_runs(
self,
conn: sqlite3.Connection,
observed_at: int,
runs: list[tuple[str, int]],
expected_rows: list[tuple[int]],
) -> None:
"""An ok and error run sharing observed_at expose only the ok run's rows."""
"""Snapshot views join same-second rows by run_id."""
create_snapshot_tables(conn)
run_err = start_snapshot_run(conn, 3000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 99)",
(run_err,),
)
record_snapshot_run(conn, run_err, "error")
run_ok = start_snapshot_run(conn, 3000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 12345)",
(run_ok,),
)
record_snapshot_run(conn, run_ok, "ok")
for status, login in runs:
run_id = start_snapshot_run(conn, observed_at)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, ?)",
(run_id, login),
)
record_snapshot_run(conn, run_id, status)
create_grafana_views(conn)
rows = conn.execute("SELECT login FROM grafana_account_snapshots").fetchall()
assert rows == [(12345,)]
def test_snapshot_view_two_ok_runs_same_second_no_duplication(
self,
conn: sqlite3.Connection,
) -> None:
"""Two ok runs sharing observed_at each produce exactly one row in the view."""
create_snapshot_tables(conn)
run1 = start_snapshot_run(conn, 4000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 1)",
(run1,),
)
record_snapshot_run(conn, run1, "ok")
run2 = start_snapshot_run(conn, 4000)
conn.execute(
"INSERT INTO account_snapshots (run_id, login) VALUES (?, 2)",
(run2,),
)
record_snapshot_run(conn, run2, "ok")
create_grafana_views(conn)
rows = conn.execute("SELECT login FROM grafana_account_snapshots").fetchall()
assert len(rows) == 2
assert rows == expected_rows
# ---------------------------------------------------------------------------
@@ -569,45 +518,32 @@ class TestGrafanaIndexes:
assert "idx_order_snapshots_time_symbol" not in indexes
assert "idx_snapshot_runs_time_status" not in indexes
def test_rates_index_skipped_when_cols_missing(
@pytest.mark.parametrize(
("ddl", "index_name"),
[
("CREATE TABLE rates (open REAL)", "idx_rates_time_symbol_timeframe"),
("CREATE TABLE ticks (bid REAL)", "idx_ticks_time_symbol"),
(
"CREATE TABLE history_deals (ticket INTEGER)",
"idx_history_deals_time_symbol",
),
(
"CREATE TABLE history_orders (ticket INTEGER)",
"idx_history_orders_time_setup_symbol",
),
],
ids=["rates", "ticks", "deals", "orders"],
)
def test_index_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
ddl: str,
index_name: str,
) -> None:
"""Rates index is skipped when required columns are absent."""
conn.execute("CREATE TABLE rates (open REAL)")
"""An index is skipped when the source table lacks required columns."""
conn.execute(ddl)
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_rates_time_symbol_timeframe" not in indexes
def test_ticks_index_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""Ticks index is skipped when required columns are absent."""
conn.execute("CREATE TABLE ticks (bid REAL)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_ticks_time_symbol" not in indexes
def test_deals_indexes_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""history_deals indexes are skipped when required columns are absent."""
conn.execute("CREATE TABLE history_deals (ticket INTEGER)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_history_deals_time_symbol" not in indexes
def test_orders_index_skipped_when_cols_missing(
self,
conn: sqlite3.Connection,
) -> None:
"""history_orders index is skipped when required columns are absent."""
conn.execute("CREATE TABLE history_orders (ticket INTEGER)")
create_grafana_indexes(conn)
indexes = _get_names(conn, "index")
assert "idx_history_orders_time_setup_symbol" not in indexes
assert index_name not in _get_names(conn, "index")
def test_snapshot_indexes_skipped_when_cols_missing(
self,
@@ -678,25 +614,56 @@ class TestSnapshotInserts:
"""Create snapshot tables before each insert test."""
create_snapshot_tables(conn)
def test_insert_account_snapshot(self, conn: sqlite3.Connection) -> None:
"""insert_account_snapshot appends a row with correct values."""
@pytest.mark.parametrize(
("insert_func", "row", "select_sql", "expected"),
[
(
insert_account_snapshot,
{
"login": 12345,
"currency": "USD",
"balance": 10000.0,
"equity": 9800.0,
"margin": 200.0,
"margin_free": 9800.0,
"margin_level": 4900.0,
"profit": -200.0,
"leverage": 100,
},
"SELECT login, currency, balance FROM account_snapshots",
(12345, "USD", 10000.0),
),
(
insert_terminal_snapshot,
{
"name": "MetaTrader 5",
"connected": 1,
"community_account": 0,
"trade_allowed": 1,
"trade_expert": 1,
"path": "/mt5",
"company": "Broker",
"language": "en",
},
"SELECT name, connected FROM terminal_snapshots",
("MetaTrader 5", 1),
),
],
ids=["account", "terminal"],
)
def test_insert_single_snapshot(
self,
conn: sqlite3.Connection,
insert_func: Callable[[sqlite3.Connection, int, dict[str, object]], None],
row: dict[str, object],
select_sql: str,
expected: tuple[object, ...],
) -> None:
"""insert_account_snapshot and insert_terminal_snapshot append one row."""
run_id = start_snapshot_run(conn, 1700000000)
row: dict[str, object] = {
"login": 12345,
"currency": "USD",
"balance": 10000.0,
"equity": 9800.0,
"margin": 200.0,
"margin_free": 9800.0,
"margin_level": 4900.0,
"profit": -200.0,
"leverage": 100,
}
insert_account_snapshot(conn, run_id, row)
result = conn.execute(
"SELECT login, currency, balance FROM account_snapshots"
).fetchone()
assert result == (12345, "USD", 10000.0)
insert_func(conn, run_id, row)
result = conn.execute(select_sql).fetchone()
assert result == expected
def test_insert_account_snapshot_partial_row(
self,
@@ -710,105 +677,91 @@ class TestSnapshotInserts:
).fetchone()
assert result == (1, None)
def test_insert_position_snapshots_with_rows(
@pytest.mark.parametrize(
("insert_func", "table", "rows", "expected_count"),
[
(
insert_position_snapshots,
"position_snapshots",
[
{"ticket": 1, "symbol": "EURUSD", "volume": 0.1, "profit": 10.0},
{"ticket": 2, "symbol": "GBPUSD", "volume": 0.2, "profit": -5.0},
],
2,
),
(
insert_position_snapshots,
"position_snapshots",
[],
0,
),
(
insert_order_snapshots,
"order_snapshots",
[
{
"ticket": 10,
"symbol": "EURUSD",
"type": 2,
"volume_current": 0.1,
},
],
1,
),
(
insert_order_snapshots,
"order_snapshots",
[],
0,
),
],
ids=[
"positions-with-rows",
"positions-empty-noop",
"orders-with-rows",
"orders-empty-noop",
],
)
def test_insert_snapshot_rows(
self,
conn: sqlite3.Connection,
insert_func: Callable[
[sqlite3.Connection, int, int | None, list[dict[str, object]]],
None,
],
table: str,
rows: list[dict[str, object]],
expected_count: int,
) -> None:
"""insert_position_snapshots appends each position row."""
"""insert_*_snapshots appends each row and is a no-op when empty."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [
{"ticket": 1, "symbol": "EURUSD", "volume": 0.1, "profit": 10.0},
{"ticket": 2, "symbol": "GBPUSD", "volume": 0.2, "profit": -5.0},
]
insert_position_snapshots(conn, run_id, 12345, rows)
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[0]
assert count == 2
insert_func(conn, run_id, 12345, rows)
count = conn.execute(
f"SELECT COUNT(*) FROM {table}" # noqa: S608
).fetchone()[0]
assert count == expected_count
def test_insert_position_snapshots_noop_when_empty(
@pytest.mark.parametrize(
("time_setup", "expected_stored"),
[
(_TIMESTAMP_TIME_SETUP, int(_TIMESTAMP_TIME_SETUP.timestamp())),
(1705314600, 1705314600),
("not_a_time", None),
],
ids=["timestamp", "int", "unknown-string"],
)
def test_insert_order_snapshots_normalizes_time_setup(
self,
conn: sqlite3.Connection,
time_setup: object,
expected_stored: int | None,
) -> None:
"""insert_position_snapshots is a no-op when rows is empty."""
"""insert_order_snapshots stores epoch int, int as-is, or None for unknown."""
run_id = start_snapshot_run(conn, 1700000000)
insert_position_snapshots(conn, run_id, 12345, [])
count = conn.execute("SELECT COUNT(*) FROM position_snapshots").fetchone()[0]
assert count == 0
def test_insert_order_snapshots_with_rows(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots appends each order row."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [
{"ticket": 10, "symbol": "EURUSD", "type": 2, "volume_current": 0.1},
]
insert_order_snapshots(conn, run_id, 12345, rows)
count = conn.execute("SELECT COUNT(*) FROM order_snapshots").fetchone()[0]
assert count == 1
def test_insert_order_snapshots_noop_when_empty(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots is a no-op when rows is empty."""
run_id = start_snapshot_run(conn, 1700000000)
insert_order_snapshots(conn, run_id, 12345, [])
count = conn.execute("SELECT COUNT(*) FROM order_snapshots").fetchone()[0]
assert count == 0
def test_insert_order_snapshots_normalizes_timestamp_time_setup(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots converts pd.Timestamp time_setup to epoch int."""
run_id = start_snapshot_run(conn, 1700000000)
ts = pd.Timestamp("2024-01-15 10:30:00", tz="UTC")
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": ts}]
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": time_setup}]
insert_order_snapshots(conn, run_id, 12345, rows)
stored = conn.execute("SELECT time_setup FROM order_snapshots").fetchone()[0]
assert stored == int(ts.timestamp())
def test_insert_order_snapshots_stores_int_time_setup(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots stores an integer time_setup as-is."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": 1705314600}]
insert_order_snapshots(conn, run_id, 12345, rows)
stored = conn.execute("SELECT time_setup FROM order_snapshots").fetchone()[0]
assert stored == 1705314600
def test_insert_order_snapshots_stores_null_for_unknown_time_setup_type(
self,
conn: sqlite3.Connection,
) -> None:
"""insert_order_snapshots stores NULL for an unrecognized time_setup type."""
run_id = start_snapshot_run(conn, 1700000000)
rows: list[dict[str, object]] = [{"ticket": 10, "time_setup": "not_a_time"}]
insert_order_snapshots(conn, run_id, 12345, rows)
stored = conn.execute("SELECT time_setup FROM order_snapshots").fetchone()[0]
assert stored is None
def test_insert_terminal_snapshot(self, conn: sqlite3.Connection) -> None:
"""insert_terminal_snapshot appends a terminal info row."""
run_id = start_snapshot_run(conn, 1700000000)
row: dict[str, object] = {
"name": "MetaTrader 5",
"connected": 1,
"community_account": 0,
"trade_allowed": 1,
"trade_expert": 1,
"path": "/mt5",
"company": "Broker",
"language": "en",
}
insert_terminal_snapshot(conn, run_id, row)
result = conn.execute(
"SELECT name, connected FROM terminal_snapshots"
).fetchone()
assert result == ("MetaTrader 5", 1)
assert stored == expected_stored
def test_start_snapshot_run_returns_incrementing_ids(
self,
@@ -819,25 +772,30 @@ class TestSnapshotInserts:
run2 = start_snapshot_run(conn, 1700000000)
assert run1 != run2
def test_record_snapshot_run_with_detail(
@pytest.mark.parametrize(
("status", "detail", "expected"),
[
pytest.param(
"error",
"RuntimeError: boom",
("error", "RuntimeError: boom"),
id="with-detail",
),
pytest.param("ok", None, ("ok", None), id="without-detail"),
],
)
def test_record_snapshot_run(
self,
conn: sqlite3.Connection,
status: str,
detail: str | None,
expected: tuple[str, str | None],
) -> None:
"""record_snapshot_run stores status and detail text."""
"""record_snapshot_run stores status and optional detail text."""
run_id = start_snapshot_run(conn, 1700000000)
record_snapshot_run(conn, run_id, "error", "RuntimeError: boom")
record_snapshot_run(conn, run_id, status, detail)
row = conn.execute("SELECT status, detail FROM snapshot_runs").fetchone()
assert row == ("error", "RuntimeError: boom")
def test_record_snapshot_run_without_detail(
self,
conn: sqlite3.Connection,
) -> None:
"""record_snapshot_run stores None for detail when omitted."""
run_id = start_snapshot_run(conn, 1700000000)
record_snapshot_run(conn, run_id, "ok")
row = conn.execute("SELECT status, detail FROM snapshot_runs").fetchone()
assert row == ("ok", None)
assert row == expected
# ---------------------------------------------------------------------------
@@ -859,23 +817,40 @@ def _make_source_db(path: Path) -> None:
class TestPublishGrafanaCopy:
"""Tests for publish_grafana_copy."""
def test_publish_to_fresh_target(self, tmp_path: Path) -> None:
"""publish_grafana_copy creates the target file."""
@pytest.mark.parametrize(
("target_rel", "stale_content", "required_tables"),
[
pytest.param(
Path("out") / "grafana.db",
None,
frozenset({"snapshot_runs", "account_snapshots"}),
id="fresh-target",
),
pytest.param(
Path("grafana.db"),
b"stale",
frozenset({"snapshot_runs"}),
id="overwrite-stale-target",
),
],
)
def test_publish_creates_valid_sqlite_target(
self,
tmp_path: Path,
target_rel: Path,
stale_content: bytes | None,
required_tables: frozenset[str],
) -> None:
"""publish_grafana_copy creates or replaces a valid SQLite target."""
source = tmp_path / "src.db"
target = tmp_path / "out" / "grafana.db"
target = tmp_path / target_rel
_make_source_db(source)
if stale_content is not None:
target.write_bytes(stale_content)
result = publish_grafana_copy(source, target)
assert target.exists()
assert isinstance(result, Path)
assert result == target.resolve()
def test_overwrite_existing_target(self, tmp_path: Path) -> None:
"""publish_grafana_copy replaces an existing target without error."""
source = tmp_path / "src.db"
target = tmp_path / "grafana.db"
_make_source_db(source)
target.write_bytes(b"stale")
publish_grafana_copy(source, target)
# Target must now be a valid SQLite file from source
with sqlite3.connect(target) as conn:
tables = {
row[0]
@@ -883,22 +858,7 @@ class TestPublishGrafanaCopy:
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
assert "snapshot_runs" in tables
def test_target_contains_source_tables(self, tmp_path: Path) -> None:
"""Published target contains the same tables as the source."""
source = tmp_path / "src.db"
target = tmp_path / "grafana.db"
_make_source_db(source)
publish_grafana_copy(source, target)
with sqlite3.connect(target) as conn:
tables = {
row[0]
for row in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()
}
assert {"snapshot_runs", "account_snapshots"}.issubset(tables)
assert required_tables <= tables
def test_target_can_be_opened_readonly(self, tmp_path: Path) -> None:
"""Published target can be opened with uri=True in read-only mode."""
@@ -956,14 +916,6 @@ class TestPublishGrafanaCopy:
tmp_files = list(tmp_path.glob("grafana.db.*.tmp"))
assert not tmp_files, "Temp file should be cleaned up on failure"
def test_returns_path_object(self, tmp_path: Path) -> None:
"""publish_grafana_copy returns a Path instance."""
source = tmp_path / "src.db"
target = tmp_path / "grafana.db"
_make_source_db(source)
result = publish_grafana_copy(source, target)
assert isinstance(result, Path)
def test_fresh_target_has_readable_permissions(self, tmp_path: Path) -> None:
"""Published copy is readable by the owner."""
import stat as _stat # noqa: PLC0415
tests/test_history.py
+527 -358
View File
File diff suppressed because it is too large Load Diff
tests/test_sdk.py
+595 -552
View File
File diff suppressed because it is too large Load Diff
tests/test_telemetry.py
+163 -100
View File
@@ -20,23 +20,13 @@ from mt5cli.telemetry import (
class TestNoOp:
"""Tests for _NoOp no-op instrument."""
def test_add_is_noop(self) -> None:
"""_NoOp.add accepts amount and optional attributes without error."""
@pytest.mark.parametrize("method", ["add", "set", "record"])
def test_method_is_noop(self, method: str) -> None:
"""_NoOp methods accept amount and optional attributes without error."""
noop = _NoOp()
noop.add(1.0)
noop.add(1.0, {"key": "val"})
def test_set_is_noop(self) -> None:
"""_NoOp.set accepts amount and optional attributes without error."""
noop = _NoOp()
noop.set(2.0)
noop.set(2.0, {"key": "val"})
def test_record_is_noop(self) -> None:
"""_NoOp.record accepts amount and optional attributes without error."""
noop = _NoOp()
noop.record(3.0)
noop.record(3.0, {"key": "val"})
bound = getattr(noop, method)
bound(1.0)
bound(1.0, {"key": "val"})
class TestMt5Metrics:
@@ -64,32 +54,105 @@ class TestMt5Metrics:
assert meter.create_counter.called
assert meter.create_gauge.called
def test_record_history_update_success(self) -> None:
"""record_history_update records duration and timestamp on success."""
@pytest.mark.parametrize(
(
"method",
"kwargs",
"duration_attr",
"failures_attr",
"last_success_attr",
),
[
pytest.param(
"record_history_update",
{"dataset": "rates"},
"_history_duration",
"_history_failures",
"_last_successful_update",
id="history-update",
),
pytest.param(
"record_snapshot_update",
{},
"_snapshot_duration",
"_snapshot_failures",
None,
id="snapshot-update",
),
],
)
def test_record_update_success(
self,
method: str,
kwargs: dict[str, str],
duration_attr: str,
failures_attr: str,
last_success_attr: str | None,
) -> None:
"""record_*_update records duration and timestamp on success."""
meter = MagicMock()
m = _Mt5Metrics()
m.configure(meter)
with m.record_history_update(dataset="rates"):
with getattr(m, method)(**kwargs):
pass
m._history_duration.record.assert_called_once() # type: ignore[reportPrivateUsage]
m._last_successful_update.set.assert_called_once() # type: ignore[reportPrivateUsage]
m._history_failures.add.assert_not_called() # type: ignore[reportPrivateUsage]
getattr(m, duration_attr).record.assert_called_once() # type: ignore[reportPrivateUsage]
getattr(m, failures_attr).add.assert_not_called() # type: ignore[reportPrivateUsage]
if last_success_attr is not None:
getattr(m, last_success_attr).set.assert_called_once() # type: ignore[reportPrivateUsage]
def test_record_history_update_failure(self) -> None:
"""record_history_update increments failure counter and re-raises on error."""
@pytest.mark.parametrize(
(
"method",
"kwargs",
"exc",
"duration_attr",
"failures_attr",
"failure_labels",
),
[
pytest.param(
"record_history_update",
{"dataset": "rates"},
ValueError("boom"),
"_history_duration",
"_history_failures",
{"dataset": "rates"},
id="history-update",
),
pytest.param(
"record_snapshot_update",
{},
RuntimeError("snap fail"),
"_snapshot_duration",
"_snapshot_failures",
{},
id="snapshot-update",
),
],
)
def test_record_update_failure(
self,
method: str,
kwargs: dict[str, str],
exc: BaseException,
duration_attr: str,
failures_attr: str,
failure_labels: dict[str, str],
) -> None:
"""record_*_update increments failure counter and re-raises on error."""
meter = MagicMock()
m = _Mt5Metrics()
m.configure(meter)
exc = ValueError("boom")
with (
pytest.raises(ValueError, match="boom"),
m.record_history_update(dataset="rates"),
pytest.raises(type(exc), match=str(exc)),
getattr(m, method)(**kwargs),
):
raise exc
m._history_failures.add.assert_called_once_with( # type: ignore[reportPrivateUsage]
1, {"dataset": "rates"}
getattr(m, failures_attr).add.assert_called_once_with( # type: ignore[reportPrivateUsage]
1,
failure_labels,
)
m._history_duration.record.assert_not_called() # type: ignore[reportPrivateUsage]
getattr(m, duration_attr).record.assert_not_called() # type: ignore[reportPrivateUsage]
def test_add_history_rows(self) -> None:
"""add_history_rows increments the rows-written counter."""
@@ -101,82 +164,82 @@ class TestMt5Metrics:
42, {"dataset": "rates"}
)
def test_record_snapshot_update_success(self) -> None:
"""record_snapshot_update records duration on success."""
@pytest.mark.parametrize(
("method", "kwargs", "gauge_attr", "expected_set_count"),
[
pytest.param(
"record_position_state",
{
"login": "42",
"server": "demo",
"symbol": "EURUSD",
"profit": 12.5,
"volume": 0.01,
},
"_position_profit",
2,
id="position-state",
),
pytest.param(
"record_terminal_state",
{
"connected": 1.0,
"trade_allowed": 1.0,
"trade_expert": 0.0,
},
"_terminal_connected",
3,
id="terminal-state",
),
pytest.param(
"record_account_state",
{
"login": "99",
"server": "live",
"balance": 5000.0,
"equity": 5100.0,
"margin": 200.0,
"margin_free": 4800.0,
"margin_level": 2550.0,
},
"_account_balance",
5,
id="account-state",
),
],
)
def test_record_state_emits_gauges(
self,
method: str,
kwargs: dict[str, float | str],
gauge_attr: str,
expected_set_count: int,
) -> None:
"""record_*_state emits the expected gauge set calls after configure."""
meter = MagicMock()
m = _Mt5Metrics()
m.configure(meter)
with m.record_snapshot_update():
pass
m._snapshot_duration.record.assert_called_once() # type: ignore[reportPrivateUsage]
m._snapshot_failures.add.assert_not_called() # type: ignore[reportPrivateUsage]
def test_record_snapshot_update_failure(self) -> None:
"""record_snapshot_update increments failure counter and re-raises on error."""
meter = MagicMock()
m = _Mt5Metrics()
m.configure(meter)
exc = RuntimeError("snap fail")
with (
pytest.raises(RuntimeError, match="snap fail"),
m.record_snapshot_update(),
):
raise exc
m._snapshot_failures.add.assert_called_once_with(1, {}) # type: ignore[reportPrivateUsage]
m._snapshot_duration.record.assert_not_called() # type: ignore[reportPrivateUsage]
def test_record_position_state(self) -> None:
"""record_position_state emits profit and volume gauges."""
meter = MagicMock()
m = _Mt5Metrics()
m.configure(meter)
m.record_position_state(
login="42",
server="demo",
symbol="EURUSD",
profit=12.5,
volume=0.01,
getattr(m, method)(**kwargs)
# Related gauges share the same create_gauge mock; verify total set calls.
assert ( # type: ignore[reportPrivateUsage]
getattr(m, gauge_attr).set.call_count == expected_set_count
)
# Both profit and volume share the same gauge mock via create_gauge.
# Verify that set was called exactly twice (once each).
assert m._position_profit.set.call_count == 2 # type: ignore[reportPrivateUsage]
def test_record_terminal_state(self) -> None:
"""record_terminal_state emits connected, trade_allowed, trade_expert gauges."""
meter = MagicMock()
@pytest.mark.parametrize(
("method", "kwargs"),
[
("record_history_update", {"dataset": "ticks"}),
("record_snapshot_update", {}),
],
)
def test_record_update_noop_before_configure(
self,
method: str,
kwargs: dict[str, str],
) -> None:
"""record_*_update works without configure (no-op instruments)."""
m = _Mt5Metrics()
m.configure(meter)
m.record_terminal_state(connected=1.0, trade_allowed=1.0, trade_expert=0.0)
# All three terminal gauges share the same mock; set is called 3 times.
assert m._terminal_connected.set.call_count == 3 # type: ignore[reportPrivateUsage]
def test_record_account_state_after_configure(self) -> None:
"""record_account_state emits all five account gauges."""
meter = MagicMock()
m = _Mt5Metrics()
m.configure(meter)
m.record_account_state(
login="99",
server="live",
balance=5000.0,
equity=5100.0,
margin=200.0,
margin_free=4800.0,
margin_level=2550.0,
)
# All five account gauges share the same gauge mock; set is called 5 times.
assert m._account_balance.set.call_count == 5 # type: ignore[reportPrivateUsage]
def test_record_history_update_noop_before_configure(self) -> None:
"""record_history_update works without configure (no-op instruments)."""
m = _Mt5Metrics()
with m.record_history_update(dataset="ticks"):
pass
def test_record_snapshot_update_noop_before_configure(self) -> None:
"""record_snapshot_update works without configure (no-op instruments)."""
m = _Mt5Metrics()
with m.record_snapshot_update():
with getattr(m, method)(**kwargs):
pass
tests/test_trading.py
+553 -707
View File
File diff suppressed because it is too large Load Diff
tests/test_utils.py
+167 -144
View File
@@ -89,28 +89,45 @@ class TestExportDataframe:
"""Create a sample DataFrame for testing."""
return pd.DataFrame({"a": [1, 2, 3], "b": ["x", "y", "z"]})
def test_export_csv(self, tmp_path: Path, sample_df: pd.DataFrame) -> None:
"""Test CSV export."""
output = tmp_path / "out.csv"
export_dataframe(sample_df, output, "csv")
result = pd.read_csv(output)
pd.testing.assert_frame_equal(result, sample_df)
def test_export_json(self, tmp_path: Path, sample_df: pd.DataFrame) -> None:
"""Test JSON export."""
output = tmp_path / "out.json"
export_dataframe(sample_df, output, "json")
with output.open() as f:
records = json.load(f)
assert len(records) == 3
assert records[0]["a"] == 1
def test_export_parquet(self, tmp_path: Path, sample_df: pd.DataFrame) -> None:
"""Test Parquet export."""
output = tmp_path / "out.parquet"
export_dataframe(sample_df, output, "parquet")
result = pd.read_parquet(output)
pd.testing.assert_frame_equal(result, sample_df)
@pytest.mark.parametrize(
("filename", "output_format", "reader"),
[
("out.csv", "csv", "csv"),
("out.json", "json", "json"),
("out.parquet", "parquet", "parquet"),
("out.db", "sqlite3", "sqlite3"),
],
ids=["csv", "json", "parquet", "sqlite3"],
)
def test_export_round_trip(
self,
tmp_path: Path,
sample_df: pd.DataFrame,
filename: str,
output_format: str,
reader: str,
) -> None:
"""Test CSV/JSON/Parquet/SQLite3 exports round-trip the sample DataFrame."""
output = tmp_path / filename
export_dataframe(sample_df, output, output_format, table_name="test_table")
if reader == "csv":
result = pd.read_csv(output)
pd.testing.assert_frame_equal(result, sample_df)
elif reader == "json":
with output.open() as f:
records = json.load(f)
assert len(records) == 3
assert records[0]["a"] == 1
elif reader == "sqlite3":
with sqlite3.connect(output) as conn:
result = pd.read_sql( # type: ignore[reportUnknownMemberType]
"SELECT * FROM test_table",
conn,
)
pd.testing.assert_frame_equal(result, sample_df)
else:
result = pd.read_parquet(output)
pd.testing.assert_frame_equal(result, sample_df)
def test_export_parquet_without_pyarrow(
self,
@@ -123,17 +140,6 @@ class TestExportDataframe:
with pytest.raises(ImportError, match="mt5cli\\[parquet\\]"):
export_dataframe(sample_df, tmp_path / "out.parquet", "parquet")
def test_export_sqlite3(self, tmp_path: Path, sample_df: pd.DataFrame) -> None:
"""Test SQLite3 export."""
output = tmp_path / "out.db"
export_dataframe(sample_df, output, "sqlite3", table_name="test_table")
with sqlite3.connect(output) as conn:
result = pd.read_sql( # type: ignore[reportUnknownMemberType]
"SELECT * FROM test_table",
conn,
)
pd.testing.assert_frame_equal(result, sample_df)
def test_unsupported_format_raises(
self,
tmp_path: Path,
@@ -147,13 +153,41 @@ class TestExportDataframe:
class TestExportDataframeToSqlite:
"""Tests for export_dataframe_to_sqlite."""
def test_append_preserves_existing_rows(self, tmp_path: Path) -> None:
"""Test append mode keeps prior rows in the SQLite table."""
@pytest.mark.parametrize(
("first_if_exists", "second_if_exists"),
[
pytest.param(IfExists.REPLACE, IfExists.APPEND, id="replace-then-append"),
pytest.param(None, None, id="default-append"),
],
)
def test_append_preserves_existing_rows(
self,
tmp_path: Path,
first_if_exists: IfExists | None,
second_if_exists: IfExists | None,
) -> None:
"""Test explicit append and default append modes keep prior rows."""
output = tmp_path / "append.db"
first = pd.DataFrame({"id": [1], "value": ["a"]})
second = pd.DataFrame({"id": [2], "value": ["b"]})
export_dataframe_to_sqlite(first, output, "items", if_exists=IfExists.REPLACE)
export_dataframe_to_sqlite(second, output, "items", if_exists=IfExists.APPEND)
if first_if_exists is None:
export_dataframe_to_sqlite(first, output, "items")
else:
export_dataframe_to_sqlite(
first,
output,
"items",
if_exists=first_if_exists,
)
if second_if_exists is None:
export_dataframe_to_sqlite(second, output, "items")
else:
export_dataframe_to_sqlite(
second,
output,
"items",
if_exists=second_if_exists,
)
with sqlite3.connect(output) as conn:
result = pd.read_sql( # type: ignore[reportUnknownMemberType]
"SELECT id, value FROM items ORDER BY id",
@@ -205,26 +239,6 @@ class TestExportDataframeToSqlite:
}),
)
def test_default_if_exists_appends_without_dropping_rows(
self,
tmp_path: Path,
) -> None:
"""Test the default append mode keeps prior rows."""
output = tmp_path / "default-append.db"
first = pd.DataFrame({"id": [1], "value": ["a"]})
second = pd.DataFrame({"id": [2], "value": ["b"]})
export_dataframe_to_sqlite(first, output, "items")
export_dataframe_to_sqlite(second, output, "items")
with sqlite3.connect(output) as conn:
result = pd.read_sql( # type: ignore[reportUnknownMemberType]
"SELECT id, value FROM items ORDER BY id",
conn,
)
pd.testing.assert_frame_equal(
result,
pd.DataFrame({"id": [1, 2], "value": ["a", "b"]}),
)
def test_writes_index_with_label(self, tmp_path: Path) -> None:
"""Test optional index export with a custom label."""
output = tmp_path / "index.db"
@@ -258,15 +272,17 @@ class TestExportDataframeToSqlite:
class TestParseDatetime:
"""Tests for parse_datetime."""
def test_valid_date(self) -> None:
"""Test parsing a date string."""
result = parse_datetime("2024-01-15")
assert result == datetime(2024, 1, 15, tzinfo=UTC)
def test_valid_datetime_with_tz(self) -> None:
"""Test parsing a datetime with timezone."""
result = parse_datetime("2024-01-15T12:00:00+00:00")
assert result == datetime(2024, 1, 15, 12, 0, 0, tzinfo=UTC)
@pytest.mark.parametrize(
("value", "expected"),
[
("2024-01-15", datetime(2024, 1, 15, tzinfo=UTC)),
("2024-01-15T12:00:00+00:00", datetime(2024, 1, 15, 12, 0, 0, tzinfo=UTC)),
],
ids=["date", "datetime-with-tz"],
)
def test_valid_inputs(self, value: str, expected: datetime) -> None:
"""Test parsing valid date and datetime strings."""
assert parse_datetime(value) == expected
def test_invalid_format_raises(self) -> None:
"""Test that invalid format raises ValueError."""
@@ -279,26 +295,29 @@ class TestParseTimeframe:
@pytest.mark.parametrize(
("value", "expected"),
[("M1", 1), ("h1", 16385), ("D1", 16408), ("MN1", 49153)],
[
("M1", 1),
("h1", 16385),
("D1", 16408),
("MN1", 49153),
("1", 1),
(16385, 16385),
],
ids=["M1", "h1", "D1", "MN1", "int-string-1", "int-16385"],
)
def test_named_timeframe(self, value: str, expected: int) -> None:
"""Test parsing named timeframes."""
def test_valid_timeframe(self, value: str | int, expected: int) -> None:
"""Test parsing valid string and integer timeframe values."""
assert parse_timeframe(value) == expected
def test_integer_timeframe(self) -> None:
"""Test parsing supported integer timeframes."""
assert parse_timeframe("1") == 1
assert parse_timeframe(16385) == 16385
def test_unsupported_integer_timeframe_raises(self) -> None:
"""Test that unsupported integer timeframes raise ValueError."""
@pytest.mark.parametrize(
"value",
["42", "INVALID"],
ids=["unsupported-integer", "invalid-string"],
)
def test_invalid_timeframe_raises(self, value: str) -> None:
"""Test that invalid timeframe values raise ValueError."""
with pytest.raises(ValueError, match="Invalid timeframe"):
parse_timeframe("42")
def test_invalid_timeframe_raises(self) -> None:
"""Test that invalid timeframe raises ValueError."""
with pytest.raises(ValueError, match="Invalid timeframe"):
parse_timeframe("INVALID")
parse_timeframe(value)
class TestParseTickFlags:
@@ -306,26 +325,29 @@ class TestParseTickFlags:
@pytest.mark.parametrize(
("value", "expected"),
[("ALL", -1), ("info", 1), ("TRADE", 2), ("COPY_TICKS_ALL", -1)],
[
("ALL", -1),
("info", 1),
("TRADE", 2),
("COPY_TICKS_ALL", -1),
("-1", -1),
(2, 2),
],
ids=["ALL", "info", "TRADE", "COPY_TICKS_ALL", "int-string--1", "int-2"],
)
def test_named_flag(self, value: str, expected: int) -> None:
"""Test parsing named tick flags."""
def test_valid_flag(self, value: str | int, expected: int) -> None:
"""Test parsing valid string and integer tick flag values."""
assert parse_tick_flags(value) == expected
def test_integer_flag(self) -> None:
"""Test parsing supported integer tick flags."""
assert parse_tick_flags("-1") == -1
assert parse_tick_flags(2) == 2
def test_unsupported_integer_flag_raises(self) -> None:
"""Test that unsupported integer tick flags raise ValueError."""
@pytest.mark.parametrize(
"value",
["7", "INVALID"],
ids=["unsupported-integer", "invalid-string"],
)
def test_invalid_flag_raises(self, value: str) -> None:
"""Test that invalid tick flag values raise ValueError."""
with pytest.raises(ValueError, match="Invalid tick flags"):
parse_tick_flags("7")
def test_invalid_flag_raises(self) -> None:
"""Test that invalid flag raises ValueError."""
with pytest.raises(ValueError, match="Invalid tick flags"):
parse_tick_flags("INVALID")
parse_tick_flags(value)
# ---------------------------------------------------------------------------
@@ -348,15 +370,17 @@ class TestParseRequest:
result = parse_request(f"@{path}")
assert result == {"action": 2}
def test_invalid_json_raises(self) -> None:
"""Test that invalid JSON raises ValueError."""
with pytest.raises(ValueError, match="Invalid JSON request"):
parse_request("not json")
def test_non_object_raises(self) -> None:
"""Test that a non-object JSON raises ValueError."""
with pytest.raises(ValueError, match="must be a JSON object"):
parse_request("[1, 2, 3]")
@pytest.mark.parametrize(
("value", "match"),
[
pytest.param("not json", "Invalid JSON request", id="invalid-json"),
pytest.param("[1, 2, 3]", "must be a JSON object", id="non-object"),
],
)
def test_invalid_request_raises(self, value: str, match: str) -> None:
"""Test that invalid JSON and non-object requests raise ValueError."""
with pytest.raises(ValueError, match=match):
parse_request(value)
def test_missing_file_raises(self, tmp_path: Path) -> None:
"""Test that a missing request file raises ValueError."""
@@ -373,13 +397,10 @@ class TestParseRequest:
class TestConstants:
"""Tests for module constants."""
def test_timeframe_map_is_private_in_utils(self) -> None:
"""TIMEFRAME_MAP is a private implementation detail; not a public attribute."""
assert not hasattr(mt5cli.utils, "TIMEFRAME_MAP")
def test_tick_flag_map_absent_from_utils(self) -> None:
"""TICK_FLAG_MAP is not exposed by mt5cli.utils."""
assert not hasattr(mt5cli.utils, "TICK_FLAG_MAP")
@pytest.mark.parametrize("name", ["TIMEFRAME_MAP", "TICK_FLAG_MAP"])
def test_private_maps_absent_from_utils(self, name: str) -> None:
"""Private pdmt5 maps are not exposed as public mt5cli.utils attributes."""
assert not hasattr(mt5cli.utils, name)
@pytest.mark.parametrize(
("dataset", "expected"),
@@ -422,23 +443,24 @@ class TestDateTimeType:
class TestTimeframeType:
"""Tests for _TimeframeType."""
def test_convert_string(self) -> None:
"""Test converting a string to timeframe integer."""
assert TIMEFRAME_TYPE.convert("H1", None, None) == 16385
@pytest.mark.parametrize(
("value", "expected"),
[("H1", 16385), (16385, 16385)],
ids=["string", "int"],
)
def test_convert_valid(self, value: str | int, expected: int) -> None:
"""Test converting valid string and integer timeframe values."""
assert TIMEFRAME_TYPE.convert(value, None, None) == expected
def test_convert_int(self) -> None:
"""Test converting supported integer timeframe values."""
assert TIMEFRAME_TYPE.convert(16385, None, None) == 16385
def test_convert_unsupported_int(self) -> None:
"""Test that unsupported integer values raise BadParameter."""
@pytest.mark.parametrize(
"value",
[42, "bad"],
ids=["unsupported-int", "invalid-string"],
)
def test_convert_invalid(self, value: object) -> None:
"""Test that unsupported int and invalid string values raise BadParameter."""
with pytest.raises(Exception, match="Invalid timeframe"):
TIMEFRAME_TYPE.convert(42, None, None)
def test_convert_invalid(self) -> None:
"""Test that invalid values raise BadParameter."""
with pytest.raises(Exception, match="Invalid timeframe"):
TIMEFRAME_TYPE.convert("bad", None, None)
TIMEFRAME_TYPE.convert(value, None, None)
@pytest.mark.parametrize("value", [True, False, None, 1.5])
def test_convert_invalid_types(self, value: object) -> None:
@@ -450,18 +472,24 @@ class TestTimeframeType:
class TestTickFlagsType:
"""Tests for _TickFlagsType."""
def test_convert_string(self) -> None:
"""Test converting a string to tick flags integer."""
assert TICK_FLAGS_TYPE.convert("ALL", None, None) == -1
@pytest.mark.parametrize(
("value", "expected"),
[("ALL", -1), (2, 2)],
ids=["string", "int"],
)
def test_convert_valid(self, value: str | int, expected: int) -> None:
"""Test converting valid string and integer tick flag values."""
assert TICK_FLAGS_TYPE.convert(value, None, None) == expected
def test_convert_int(self) -> None:
"""Test converting supported integer tick flag values."""
assert TICK_FLAGS_TYPE.convert(2, None, None) == 2
def test_convert_unsupported_int(self) -> None:
"""Test that unsupported integer values raise BadParameter."""
@pytest.mark.parametrize(
"value",
[7, "bad"],
ids=["unsupported-int", "invalid-string"],
)
def test_convert_invalid(self, value: object) -> None:
"""Test that unsupported int and invalid string values raise BadParameter."""
with pytest.raises(Exception, match="Invalid tick flags"):
TICK_FLAGS_TYPE.convert(7, None, None)
TICK_FLAGS_TYPE.convert(value, None, None)
@pytest.mark.parametrize("value", [True, False, None, 1.5])
def test_convert_invalid_types(self, value: object) -> None:
@@ -469,11 +497,6 @@ class TestTickFlagsType:
with pytest.raises(Exception, match="Invalid tick flags"):
TICK_FLAGS_TYPE.convert(value, None, None)
def test_convert_invalid(self) -> None:
"""Test that invalid values raise BadParameter."""
with pytest.raises(Exception, match="Invalid tick flags"):
TICK_FLAGS_TYPE.convert("bad", None, None)
class TestRequestType:
"""Tests for _RequestType."""