Tacit Lab
62c40a9776
- Fix TOCTOU race conditions by wrapping read-modify-write cycles under single-file locks in execution_state, portfolio_service, precheck_state, state_manager, and precheck_service. - Add missing test coverage (96 tests total): - test_review_service.py (15 tests) - test_check_api.py (6 tests) - test_external_gate.py main branches (+10 tests) - test_trade_execution.py new commands (+8 tests) - Unify all agent-consumed JSON messages to English. - Config-ize hardcoded values (volume filter, schema_version) via get_user_config with sensible defaults. - Add 1-hour TTL to exchange cache with force_new override. - Add ruff and mypy to dev dependencies; fix all type errors. - Add __all__ declarations to 11 service modules. - Sync README with new commands, config tuning docs, and PyPI badge. - Publish package as coinhunter==1.0.0 on PyPI with MIT license. - Deprecate coinhunter-cli==1.0.1 with runtime warning. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
204 lines
9.6 KiB
Python
204 lines
9.6 KiB
Python
"""Tests for external_gate configurable hook."""
|
|
|
|
import json
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from coinhunter.commands import external_gate
|
|
|
|
|
|
class TestResolveTriggerCommand:
|
|
def test_missing_config_means_disabled(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
# no config file exists
|
|
cmd = external_gate._resolve_trigger_command(paths)
|
|
assert cmd is None
|
|
|
|
def test_uses_explicit_list_from_config(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({
|
|
"external_gate": {"trigger_command": ["my-scheduler", "run", "job-123"]}
|
|
}), encoding="utf-8")
|
|
cmd = external_gate._resolve_trigger_command(paths)
|
|
assert cmd == ["my-scheduler", "run", "job-123"]
|
|
|
|
def test_null_means_disabled(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({
|
|
"external_gate": {"trigger_command": None}
|
|
}), encoding="utf-8")
|
|
cmd = external_gate._resolve_trigger_command(paths)
|
|
assert cmd is None
|
|
|
|
def test_empty_list_means_disabled(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({
|
|
"external_gate": {"trigger_command": []}
|
|
}), encoding="utf-8")
|
|
cmd = external_gate._resolve_trigger_command(paths)
|
|
assert cmd is None
|
|
|
|
def test_string_gets_wrapped(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({
|
|
"external_gate": {"trigger_command": "my-script.sh"}
|
|
}), encoding="utf-8")
|
|
cmd = external_gate._resolve_trigger_command(paths)
|
|
assert cmd == ["my-script.sh"]
|
|
|
|
def test_unexpected_type_returns_none_and_warns(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({
|
|
"external_gate": {"trigger_command": 42}
|
|
}), encoding="utf-8")
|
|
cmd = external_gate._resolve_trigger_command(paths)
|
|
assert cmd is None
|
|
|
|
|
|
class TestMain:
|
|
def test_already_running(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.state_dir.mkdir(parents=True, exist_ok=True)
|
|
# Acquire the lock in this process so main() sees it as busy
|
|
import fcntl
|
|
with open(paths.external_gate_lock, "w", encoding="utf-8") as f:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
|
|
rc = external_gate.main()
|
|
assert rc == 0
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "already_running"
|
|
|
|
def test_precheck_failure(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
fake_result = MagicMock(returncode=1, stdout="err", stderr="")
|
|
with patch("coinhunter.commands.external_gate.run_cmd", return_value=fake_result):
|
|
rc = external_gate.main()
|
|
assert rc == 1
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "precheck_failed"
|
|
|
|
def test_precheck_parse_error(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
fake_result = MagicMock(returncode=0, stdout="not-json", stderr="")
|
|
with patch("coinhunter.commands.external_gate.run_cmd", return_value=fake_result):
|
|
rc = external_gate.main()
|
|
assert rc == 1
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "precheck_parse_error"
|
|
|
|
def test_precheck_not_ok(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
fake_result = MagicMock(returncode=0, stdout=json.dumps({"ok": False}), stderr="")
|
|
with patch("coinhunter.commands.external_gate.run_cmd", return_value=fake_result):
|
|
rc = external_gate.main()
|
|
assert rc == 1
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "precheck_not_ok"
|
|
|
|
def test_no_trigger(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
fake_result = MagicMock(returncode=0, stdout=json.dumps({"ok": True, "should_analyze": False}), stderr="")
|
|
with patch("coinhunter.commands.external_gate.run_cmd", return_value=fake_result):
|
|
rc = external_gate.main()
|
|
assert rc == 0
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "no_trigger"
|
|
|
|
def test_already_queued(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
fake_result = MagicMock(
|
|
returncode=0,
|
|
stdout=json.dumps({"ok": True, "should_analyze": True, "run_requested": True, "run_requested_at": "2024-01-01T00:00:00Z"}),
|
|
stderr="",
|
|
)
|
|
with patch("coinhunter.commands.external_gate.run_cmd", return_value=fake_result):
|
|
rc = external_gate.main()
|
|
assert rc == 0
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "already_queued"
|
|
|
|
def test_trigger_disabled(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({"external_gate": {"trigger_command": None}}), encoding="utf-8")
|
|
# First call is precheck, second is mark-run-requested
|
|
responses = [
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True, "should_analyze": True}), stderr=""),
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True}), stderr=""),
|
|
]
|
|
with patch("coinhunter.commands.external_gate.run_cmd", side_effect=responses):
|
|
rc = external_gate.main()
|
|
assert rc == 0
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "trigger_disabled"
|
|
|
|
def test_trigger_success(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({"external_gate": {"trigger_command": ["echo", "ok"]}}), encoding="utf-8")
|
|
responses = [
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True, "should_analyze": True, "reasons": ["price-move"]}), stderr=""),
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True}), stderr=""),
|
|
MagicMock(returncode=0, stdout="triggered", stderr=""),
|
|
]
|
|
with patch("coinhunter.commands.external_gate.run_cmd", side_effect=responses):
|
|
rc = external_gate.main()
|
|
assert rc == 0
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["triggered"] is True
|
|
assert out["reason"] == "price-move"
|
|
|
|
def test_trigger_command_failure(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({"external_gate": {"trigger_command": ["false"]}}), encoding="utf-8")
|
|
responses = [
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True, "should_analyze": True}), stderr=""),
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True}), stderr=""),
|
|
MagicMock(returncode=1, stdout="", stderr="fail"),
|
|
]
|
|
with patch("coinhunter.commands.external_gate.run_cmd", side_effect=responses):
|
|
rc = external_gate.main()
|
|
assert rc == 1
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "trigger_failed"
|
|
|
|
def test_mark_run_requested_failure(self, tmp_path, monkeypatch, capsys):
|
|
monkeypatch.setenv("COINHUNTER_HOME", str(tmp_path))
|
|
paths = external_gate._paths()
|
|
paths.root.mkdir(parents=True, exist_ok=True)
|
|
paths.config_file.write_text(json.dumps({"external_gate": {"trigger_command": ["echo", "ok"]}}), encoding="utf-8")
|
|
responses = [
|
|
MagicMock(returncode=0, stdout=json.dumps({"ok": True, "should_analyze": True}), stderr=""),
|
|
MagicMock(returncode=1, stdout="err", stderr=""),
|
|
]
|
|
with patch("coinhunter.commands.external_gate.run_cmd", side_effect=responses):
|
|
rc = external_gate.main()
|
|
assert rc == 1
|
|
out = json.loads(capsys.readouterr().out)
|
|
assert out["reason"] == "mark_failed"
|