111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
"""A model error during screening is printed and fails the command."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from cli.prisma import app
|
|
|
|
runner = CliRunner()
|
|
|
|
_ERR = {"screened": 0, "include": 0, "exclude": 0, "uncertain": 0, "errors": 2}
|
|
_OK = {"screened": 0, "include": 0, "exclude": 0, "uncertain": 0, "errors": 0}
|
|
|
|
|
|
def _mock_db():
|
|
db = MagicMock()
|
|
db.__enter__ = MagicMock(return_value=db)
|
|
db.__exit__ = MagicMock(return_value=False)
|
|
return db
|
|
|
|
|
|
def _ready(mc_db, mc_load, mc_provider):
|
|
mc_db.return_value = _mock_db()
|
|
mc_load.return_value = MagicMock()
|
|
mc_provider.return_value = MagicMock()
|
|
|
|
|
|
class TestScreenErrors:
|
|
@patch("prisma.screen.run", return_value=_ERR)
|
|
@patch("prisma.llm.make_provider")
|
|
@patch("prisma.project.load")
|
|
@patch("zot.db.Db")
|
|
def test_screen_exits_1_when_the_model_errors(
|
|
self, mc_db, mc_load, mc_provider, mc_run
|
|
):
|
|
_ready(mc_db, mc_load, mc_provider)
|
|
result = runner.invoke(app, ["screen", "skin-subs", "--no-hold"])
|
|
assert result.exit_code == 1
|
|
assert "errors" in result.output
|
|
|
|
@patch("prisma.screen.run", return_value=_OK)
|
|
@patch("prisma.llm.make_provider")
|
|
@patch("prisma.project.load")
|
|
@patch("zot.db.Db")
|
|
def test_screen_exits_0_when_there_are_no_errors(
|
|
self, mc_db, mc_load, mc_provider, mc_run
|
|
):
|
|
_ready(mc_db, mc_load, mc_provider)
|
|
result = runner.invoke(app, ["screen", "skin-subs", "--no-hold"])
|
|
assert result.exit_code == 0
|
|
|
|
@patch("prisma.vpn.status", return_value={"status": "down"})
|
|
@patch("prisma.flow.text_summary", return_value="Summary: OK")
|
|
@patch("prisma.flow.count")
|
|
@patch("prisma.extract.run", return_value={"extracted": 0})
|
|
@patch("prisma.eligibility.run", return_value={"assessed": 0, "errors": 0})
|
|
@patch("prisma.fetch.run", return_value={"missed": 0, "errors": 0})
|
|
@patch("prisma.screen.run", return_value=_ERR)
|
|
@patch("prisma.llm.make_provider")
|
|
@patch("prisma.project.load")
|
|
@patch("zot.db.Db")
|
|
def test_run_exits_1_when_the_screen_stage_errors(
|
|
self,
|
|
mc_db,
|
|
mc_load,
|
|
mc_provider,
|
|
mc_screen,
|
|
mc_fetch,
|
|
mc_elig,
|
|
mc_extract,
|
|
mc_count,
|
|
mc_summary,
|
|
mc_vpn,
|
|
):
|
|
_ready(mc_db, mc_load, mc_provider)
|
|
mc_count.return_value = MagicMock()
|
|
result = runner.invoke(app, ["run", "skin-subs", "--no-hold"])
|
|
assert result.exit_code == 1
|
|
assert "errors" in result.output
|
|
mc_fetch.assert_not_called()
|
|
|
|
@patch("prisma.vpn.status", return_value={"status": "down"})
|
|
@patch("prisma.flow.text_summary", return_value="Summary: OK")
|
|
@patch("prisma.flow.count")
|
|
@patch("prisma.extract.run", return_value={"extracted": 0})
|
|
@patch("prisma.eligibility.run", return_value={"assessed": 0, "errors": 0})
|
|
@patch("prisma.fetch.run", return_value={"missed": 0, "errors": 0})
|
|
@patch("prisma.screen.run", return_value=_OK)
|
|
@patch("prisma.llm.make_provider")
|
|
@patch("prisma.project.load")
|
|
@patch("zot.db.Db")
|
|
def test_run_exits_0_when_the_screen_stage_is_clean(
|
|
self,
|
|
mc_db,
|
|
mc_load,
|
|
mc_provider,
|
|
mc_screen,
|
|
mc_fetch,
|
|
mc_elig,
|
|
mc_extract,
|
|
mc_count,
|
|
mc_summary,
|
|
mc_vpn,
|
|
):
|
|
_ready(mc_db, mc_load, mc_provider)
|
|
mc_count.return_value = MagicMock()
|
|
result = runner.invoke(app, ["run", "skin-subs", "--no-hold"])
|
|
assert result.exit_code == 0
|