651 lines
20 KiB
Python
651 lines
20 KiB
Python
import json
|
|
from pathlib import Path
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tjwater_cli import common, core
|
|
from tjwater_cli.main import app, main
|
|
|
|
|
|
runner = CliRunner()
|
|
|
|
|
|
class DummyResponse:
|
|
def __init__(self, *, status_code=200, json_data=None, text="", headers=None, content=None):
|
|
self.status_code = status_code
|
|
self._json_data = json_data
|
|
self.text = text
|
|
self.headers = headers or {"content-type": "application/json"}
|
|
self.content = content if content is not None else text.encode("utf-8")
|
|
|
|
@property
|
|
def ok(self):
|
|
return 200 <= self.status_code < 300
|
|
|
|
def json(self):
|
|
if self._json_data is None:
|
|
raise ValueError("no json")
|
|
return self._json_data
|
|
|
|
|
|
def test_load_auth_context_supports_aliases(monkeypatch):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_PROJECT_ID", "p1")
|
|
monkeypatch.setenv("TJWATER_USER_ID", "u1")
|
|
monkeypatch.setenv("TJWATER_USERNAME", "tester")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "net1")
|
|
|
|
auth = core.load_auth_context(auth_stdin=False)
|
|
|
|
assert auth.server == "http://server"
|
|
assert auth.access_token == "abc"
|
|
assert auth.project_id == "p1"
|
|
assert auth.user_id == "u1"
|
|
assert auth.username == "tester"
|
|
assert auth.network == "net1"
|
|
|
|
|
|
def test_build_runtime_context_uses_default_server(monkeypatch):
|
|
monkeypatch.delenv("TJWATER_SERVER", raising=False)
|
|
monkeypatch.delenv("TJWATER_ACCESS_TOKEN", raising=False)
|
|
monkeypatch.delenv("TJWATER_PROJECT_ID", raising=False)
|
|
monkeypatch.delenv("TJWATER_USER_ID", raising=False)
|
|
monkeypatch.delenv("TJWATER_USERNAME", raising=False)
|
|
monkeypatch.delenv("TJWATER_NETWORK", raising=False)
|
|
monkeypatch.delenv("TJWATER_EXTRA_HEADERS", raising=False)
|
|
|
|
runtime = core.build_runtime_context(
|
|
server=None,
|
|
scheme=None,
|
|
timeout=core.DEFAULT_TIMEOUT,
|
|
request_id="req-1",
|
|
)
|
|
|
|
assert runtime.server == core.DEFAULT_SERVER
|
|
|
|
|
|
def test_auth_stdin_can_be_reused_with_runtime_context_cache(monkeypatch):
|
|
observed_runtime_ids: list[int] = []
|
|
|
|
def fake_request_json(ctx, **kwargs):
|
|
observed_runtime_ids.append(id(ctx))
|
|
assert ctx.auth.access_token == "token-1"
|
|
assert kwargs["params"] == {"network": "tjwater", "node": "11"}
|
|
return {"node": "11"}, 5
|
|
|
|
monkeypatch.setattr(common, "request_json", fake_request_json)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
["--auth-stdin", "network", "get-node-properties", "--node", "11"],
|
|
input=json.dumps(
|
|
{
|
|
"server": "http://server",
|
|
"access_token": "token-1",
|
|
"project_id": "project-1",
|
|
"network": "tjwater",
|
|
}
|
|
),
|
|
)
|
|
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["ok"] is True
|
|
assert payload["data"] == {"node": "11"}
|
|
assert len(observed_runtime_ids) == 1
|
|
|
|
|
|
def test_network_get_all_junction_properties_uses_network_context(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_request_json(ctx, **kwargs):
|
|
captured["access_token"] = ctx.auth.access_token
|
|
captured["params"] = kwargs["params"]
|
|
return [{"id": "J1"}], 5
|
|
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "tjwater")
|
|
monkeypatch.setattr(common, "request_json", fake_request_json)
|
|
|
|
result = runner.invoke(app, ["network", "get-all-junction-properties"])
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["ok"] is True
|
|
assert payload["data"] == [{"id": "J1"}]
|
|
assert captured == {"access_token": "abc", "params": {"network": "tjwater"}}
|
|
|
|
|
|
def test_network_get_all_pipe_properties_uses_network_context(monkeypatch):
|
|
captured = {}
|
|
|
|
def fake_request_json(ctx, **kwargs):
|
|
captured["access_token"] = ctx.auth.access_token
|
|
captured["params"] = kwargs["params"]
|
|
return [{"id": "P1"}], 5
|
|
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "tjwater")
|
|
monkeypatch.setattr(common, "request_json", fake_request_json)
|
|
|
|
result = runner.invoke(app, ["network", "get-all-pipe-properties"])
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["ok"] is True
|
|
assert payload["data"] == [{"id": "P1"}]
|
|
assert captured == {"access_token": "abc", "params": {"network": "tjwater"}}
|
|
|
|
|
|
def test_help_outputs_json_lists_commands():
|
|
result = runner.invoke(app, ["help"])
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["schema_version"] == "tjwater-cli/v1"
|
|
assert any(command["command"] == "analysis" for command in payload["commands"])
|
|
assert all(command["command"] != "project" for command in payload["commands"])
|
|
assert payload["menu_level"] == 1
|
|
assert all(command["command"] != "project list" for command in payload["commands"])
|
|
|
|
|
|
def test_help_option_json_is_removed():
|
|
result = runner.invoke(app, ["help", "--json"])
|
|
|
|
assert result.exit_code == 2
|
|
assert "No such option: --json" in result.output
|
|
|
|
|
|
def test_simulation_help_lists_subcommands():
|
|
result = runner.invoke(app, ["simulation", "help"])
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["summary"] == "模拟运行与调度相关命令。"
|
|
commands = {command["command"]: command for command in payload["commands"]}
|
|
assert commands["simulation run"]["summary"] == "触发指定绝对时间的模拟运行"
|
|
assert commands["simulation run"]["usage"] == "tjwater-cli simulation run --start-time <START_TIME> --duration <DURATION>"
|
|
assert "tjwater-cli" in commands["simulation run"]["example"]
|
|
assert "simulation run" in commands["simulation run"]["example"]
|
|
|
|
|
|
def test_nested_group_help_lists_examples():
|
|
result = runner.invoke(app, ["analysis", "leakage", "help"])
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["summary"] == "漏损分析相关命令。"
|
|
commands = {command["command"]: command for command in payload["commands"]}
|
|
assert commands["analysis leakage identify"]["summary"] == "执行漏损识别"
|
|
assert commands["analysis leakage identify"]["example"] == "tjwater-cli analysis leakage identify --start-time 2025-01-02T03:00:00+08:00 --end-time 2025-01-02T04:00:00+08:00 --scheme leak_case_01"
|
|
|
|
|
|
def test_analysis_help_uses_group_summaries_for_nested_groups():
|
|
result = runner.invoke(app, ["analysis", "help"])
|
|
payload = json.loads(result.stdout)
|
|
commands = {command["command"]: command for command in payload["commands"]}
|
|
|
|
assert result.exit_code == 0
|
|
assert commands["analysis leakage"]["summary"] == "漏损分析相关命令。"
|
|
assert commands["analysis burst-detection"]["summary"] == "爆管检测相关命令。"
|
|
assert "analysis burst-location" not in commands
|
|
assert "analysis risk" not in commands
|
|
assert commands["analysis burst"]["example"] == "tjwater-cli analysis burst --start-time 2025-01-02T03:04:05+08:00 --duration 900 --burst-file ./burst.json --scheme burst_case_01"
|
|
assert commands["analysis valve"]["example"] == "tjwater-cli analysis valve --mode close --start-time 2025-01-02T03:04:05+08:00 --valve V1 --valve V2 --duration 900 --scheme valve_case_01"
|
|
|
|
|
|
def test_bare_analysis_uses_typer_help_with_descriptions():
|
|
result = runner.invoke(app, ["analysis"])
|
|
|
|
assert result.exit_code == 2
|
|
assert "分析计算与诊断相关命令。" in result.stdout
|
|
assert "burst 执行爆管分析" in result.stdout
|
|
assert "valve" in result.stdout
|
|
assert "leakage 漏损分析相关命令。" in result.stdout
|
|
assert "burst-location" not in result.stdout
|
|
assert "risk" not in result.stdout
|
|
|
|
|
|
def test_leaf_help_outputs_json():
|
|
result = runner.invoke(app, ["help", "simulation", "run"])
|
|
payload = json.loads(result.stdout)
|
|
|
|
assert result.exit_code == 0
|
|
assert payload["command"] == "simulation run"
|
|
assert payload["output"] == "模拟触发结果；实时数据需通过 data timeseries 命令按时间段查询"
|
|
assert payload["usage"] == "tjwater-cli simulation run --start-time <START_TIME> --duration <DURATION>"
|
|
assert len(payload["examples"]) == 1
|
|
assert "simulation run" in payload["examples"][0]
|
|
|
|
|
|
def test_root_help_flag_uses_typer_style_with_examples():
|
|
result = runner.invoke(app, ["--help"], prog_name="tjwater-cli")
|
|
|
|
assert result.exit_code == 0
|
|
assert "Usage: tjwater-cli" in result.stdout
|
|
assert "Examples:" in result.stdout
|
|
assert "tjwater-cli help simulation run" in result.stdout
|
|
|
|
|
|
def test_leaf_help_flag_includes_usage_and_example():
|
|
result = runner.invoke(app, ["simulation", "run", "--help"], prog_name="tjwater-cli")
|
|
|
|
assert result.exit_code == 0
|
|
assert "Usage: tjwater-cli simulation run [OPTIONS]" in result.stdout
|
|
assert "Usage example:" in result.stdout
|
|
assert "--start-time <START_TIME>" in result.stdout
|
|
assert "--duration" in result.stdout
|
|
assert "Examples:" in result.stdout
|
|
assert "tjwater-cli simulation run" in result.stdout
|
|
assert "START_TIME" in result.stdout
|
|
assert "DURATION" in result.stdout
|
|
|
|
|
|
def test_realtime_simulation_help_clarifies_type_values():
|
|
result = runner.invoke(
|
|
app,
|
|
["data", "timeseries", "realtime", "simulation-by-id-time", "--help"],
|
|
prog_name="tjwater-cli",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "links/nodes 是子命令" in result.stdout
|
|
assert "pipe" in result.stdout
|
|
assert "junction" in result.stdout
|
|
|
|
|
|
def test_realtime_property_help_lists_supported_fields():
|
|
result = runner.invoke(
|
|
app,
|
|
["data", "timeseries", "realtime", "simulation-by-time-property", "--help"],
|
|
prog_name="tjwater-cli",
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert "flow" in result.stdout
|
|
assert "pressure" in result.stdout
|
|
assert "actual_demand" in result.stdout
|
|
assert "velocity" in result.stdout
|
|
|
|
|
|
def test_analysis_burst_returns_next_step_to_fetch_scheme(monkeypatch, tmp_path: Path):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
burst_path = tmp_path / "burst.json"
|
|
burst_path.write_text('[{"id":"P1","size":3.5}]', encoding="utf-8")
|
|
|
|
def fake_request(**kwargs):
|
|
return DummyResponse(text="success", headers={"content-type": "text/plain"})
|
|
|
|
monkeypatch.setattr(core.requests, "request", fake_request)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"analysis",
|
|
"burst",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--duration",
|
|
"30",
|
|
"--burst-file",
|
|
str(burst_path),
|
|
"--scheme",
|
|
"burst_case_01",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert '"summary": "爆管分析执行成功"' in result.stdout
|
|
assert "tjwater-cli data scheme get --name burst_case_01" in result.stdout
|
|
assert "tjwater-cli data scheme list" in result.stdout
|
|
|
|
|
|
def test_analysis_contaminant_sends_required_scheme_name(monkeypatch):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
captured = {}
|
|
|
|
def fake_request(**kwargs):
|
|
captured.update(kwargs)
|
|
return DummyResponse(text="success", headers={"content-type": "text/plain"})
|
|
|
|
monkeypatch.setattr(core.requests, "request", fake_request)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"analysis",
|
|
"contaminant",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--duration",
|
|
"900",
|
|
"--source-node",
|
|
"N1",
|
|
"--concentration",
|
|
"10.0",
|
|
"--scheme",
|
|
"contam_case_01",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert captured["params"] == {
|
|
"network": "demo",
|
|
"start_time": "2025-01-02T03:04:05+08:00",
|
|
"source": "N1",
|
|
"concentration": 10.0,
|
|
"duration": 900,
|
|
"scheme_name": "contam_case_01",
|
|
}
|
|
|
|
|
|
def test_analysis_flushing_sends_required_scheme_name(monkeypatch, tmp_path: Path):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
captured = {}
|
|
valve_path = tmp_path / "valve.json"
|
|
valve_path.write_text('[{"valve":"V1","opening":0.5}]', encoding="utf-8")
|
|
|
|
def fake_request(**kwargs):
|
|
captured.update(kwargs)
|
|
return DummyResponse(text="success", headers={"content-type": "text/plain"})
|
|
|
|
monkeypatch.setattr(core.requests, "request", fake_request)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"analysis",
|
|
"flushing",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--valve-setting-file",
|
|
str(valve_path),
|
|
"--drainage-node",
|
|
"N1",
|
|
"--flow",
|
|
"100.0",
|
|
"--duration",
|
|
"900",
|
|
"--scheme",
|
|
"flush_case_01",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert captured["params"] == {
|
|
"network": "demo",
|
|
"start_time": "2025-01-02T03:04:05+08:00",
|
|
"valves": ["V1"],
|
|
"valves_k": [0.5],
|
|
"drainage_node_ID": "N1",
|
|
"flush_flow": 100.0,
|
|
"duration": 900,
|
|
"scheme_name": "flush_case_01",
|
|
}
|
|
|
|
|
|
def test_analysis_valve_close_sends_required_scheme_name(monkeypatch):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
captured = {}
|
|
|
|
def fake_request(**kwargs):
|
|
captured.update(kwargs)
|
|
return DummyResponse(text="success", headers={"content-type": "text/plain"})
|
|
|
|
monkeypatch.setattr(core.requests, "request", fake_request)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"analysis",
|
|
"valve",
|
|
"--mode",
|
|
"close",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--valve",
|
|
"V1",
|
|
"--duration",
|
|
"900",
|
|
"--scheme",
|
|
"valve_case_01",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert captured["params"] == {
|
|
"network": "demo",
|
|
"start_time": "2025-01-02T03:04:05+08:00",
|
|
"valves": ["V1"],
|
|
"duration": 900,
|
|
"scheme_name": "valve_case_01",
|
|
}
|
|
|
|
|
|
def test_analysis_contaminant_requires_scheme(monkeypatch, capsys):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
|
|
exit_code = main(
|
|
[
|
|
"analysis",
|
|
"contaminant",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--duration",
|
|
"900",
|
|
"--source-node",
|
|
"N1",
|
|
"--concentration",
|
|
"10.0",
|
|
],
|
|
)
|
|
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"code": "SCHEME_REQUIRED"' in stdout
|
|
|
|
|
|
def test_analysis_flushing_requires_scheme(monkeypatch, tmp_path: Path, capsys):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
valve_path = tmp_path / "valve.json"
|
|
valve_path.write_text('[{"valve":"V1","opening":0.5}]', encoding="utf-8")
|
|
|
|
exit_code = main(
|
|
[
|
|
"analysis",
|
|
"flushing",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--valve-setting-file",
|
|
str(valve_path),
|
|
"--drainage-node",
|
|
"N1",
|
|
"--flow",
|
|
"100.0",
|
|
],
|
|
)
|
|
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"code": "SCHEME_REQUIRED"' in stdout
|
|
|
|
|
|
def test_analysis_valve_close_requires_scheme(monkeypatch, capsys):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
|
|
exit_code = main(
|
|
[
|
|
"analysis",
|
|
"valve",
|
|
"--mode",
|
|
"close",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--valve",
|
|
"V1",
|
|
"--duration",
|
|
"900",
|
|
],
|
|
)
|
|
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"code": "SCHEME_REQUIRED"' in stdout
|
|
|
|
|
|
def test_main_missing_option_error_includes_usage_and_next_step(capsys):
|
|
exit_code = main(["simulation", "run"])
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"summary": "缺少参数"' in stdout
|
|
assert '"code": "MISSING_PARAMETER"' in stdout
|
|
assert '"usage": "tjwater-cli simulation run --start-time <START_TIME> --duration <DURATION>"' in stdout
|
|
assert '"tjwater-cli help simulation run"' in stdout
|
|
|
|
|
|
def test_main_invalid_enum_value_is_rejected_before_request(capsys):
|
|
exit_code = main(
|
|
[
|
|
"data",
|
|
"timeseries",
|
|
"realtime",
|
|
"simulation-by-id-time",
|
|
"--id",
|
|
"J1",
|
|
"--type",
|
|
"links",
|
|
"--time",
|
|
"2025-01-02T03:30:00+08:00",
|
|
]
|
|
)
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"summary": "参数无效"' in stdout
|
|
assert '"code": "INVALID_PARAMETER"' in stdout
|
|
assert "links" in stdout
|
|
assert "pipe" in stdout
|
|
assert "junction" in stdout
|
|
|
|
|
|
def test_main_invalid_pipe_property_is_rejected_before_request(capsys):
|
|
exit_code = main(
|
|
[
|
|
"data",
|
|
"timeseries",
|
|
"realtime",
|
|
"simulation-by-time-property",
|
|
"--type",
|
|
"pipe",
|
|
"--time",
|
|
"2025-01-02T03:30:00+08:00",
|
|
"--property",
|
|
"pressure",
|
|
]
|
|
)
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"code": "INVALID_PROPERTY"' in stdout
|
|
assert "flow" in stdout
|
|
assert "velocity" in stdout
|
|
|
|
|
|
def test_main_invalid_scada_field_is_rejected_before_request(capsys):
|
|
exit_code = main(
|
|
[
|
|
"data",
|
|
"timeseries",
|
|
"scada",
|
|
"query",
|
|
"--device-id",
|
|
"D1",
|
|
"--start-time",
|
|
"2025-01-02T03:00:00+08:00",
|
|
"--end-time",
|
|
"2025-01-02T04:00:00+08:00",
|
|
"--field",
|
|
"flow",
|
|
]
|
|
)
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"code": "INVALID_FIELD"' in stdout
|
|
assert "monitored_value" in stdout
|
|
assert "cleaned_value" in stdout
|
|
|
|
|
|
def test_main_bare_analysis_returns_typer_help_without_json_error(capsys):
|
|
exit_code = main(["analysis"])
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 0
|
|
assert "Usage: tjwater-cli analysis" in stdout
|
|
assert "分析计算与诊断相关命令。" in stdout
|
|
assert '"ok": false' not in stdout
|
|
|
|
|
|
def test_simulation_run_translates_rfc3339(monkeypatch):
|
|
monkeypatch.setenv("TJWATER_SERVER", "http://server")
|
|
monkeypatch.setenv("TJWATER_ACCESS_TOKEN", "abc")
|
|
monkeypatch.setenv("TJWATER_NETWORK", "demo")
|
|
captured = {}
|
|
|
|
def fake_request(**kwargs):
|
|
captured.update(kwargs)
|
|
return DummyResponse(json_data={"status": "success", "message": "Simulation started"})
|
|
|
|
monkeypatch.setattr(core.requests, "request", fake_request)
|
|
|
|
result = runner.invoke(
|
|
app,
|
|
[
|
|
"simulation",
|
|
"run",
|
|
"--start-time",
|
|
"2025-01-02T03:04:05+08:00",
|
|
"--duration",
|
|
"30",
|
|
],
|
|
)
|
|
|
|
assert result.exit_code == 0
|
|
assert captured["json"] == {
|
|
"name": "demo",
|
|
"start_time": "2025-01-02T03:04:05+08:00",
|
|
"duration": 30,
|
|
}
|
|
assert "tjwater-cli data timeseries realtime links" in result.stdout
|
|
assert "tjwater-cli data timeseries realtime nodes" in result.stdout
|
|
|
|
|
|
def test_removed_project_command_returns_not_found(capsys):
|
|
exit_code = main(["project", "list"])
|
|
stdout = capsys.readouterr().out
|
|
|
|
assert exit_code == 2
|
|
assert '"code": "COMMAND_NOT_FOUND"' in stdout or "No such command: project" in stdout
|