diff --git a/backend/tests/test_integration_v2.py b/backend/tests/test_integration_v2.py index fd2fef2..4c47b4c 100644 --- a/backend/tests/test_integration_v2.py +++ b/backend/tests/test_integration_v2.py @@ -115,7 +115,12 @@ for _mod in [ "passlib", "passlib.context", ]: if _mod not in sys.modules: - sys.modules[_mod] = ModuleType(_mod) + m = ModuleType(_mod) + if _mod == "boto3": + m.client = MagicMock() + elif _mod == "botocore.exceptions": + m.ClientError = Exception + sys.modules[_mod] = m # Now safe to import from app.models.enums import TestState, TestResult, TechniqueStatus diff --git a/backend/tests/test_t106_workflow_service.py b/backend/tests/test_t106_workflow_service.py index 9d32cd2..62ebdbe 100644 --- a/backend/tests/test_t106_workflow_service.py +++ b/backend/tests/test_t106_workflow_service.py @@ -1,139 +1,28 @@ """Validation tests for T-106: Test Workflow Service. -Uses mock objects to avoid needing a running database. -The database module is stubbed before any app imports. +Mocked DB sessions exercise the workflow service without PostgreSQL. +Uses the same import stack as the rest of the suite (``tests.conftest``) so +this module must **not** replace entries in ``sys.modules`` — doing so broke +JWT verification in API tests by splitting :mod:`app.auth` (loaded early via +conftest) from :mod:`app.routers.auth` / :mod:`app.dependencies.auth`. """ -import sys -import os import uuid from unittest.mock import MagicMock, patch -from types import ModuleType -from datetime import datetime - -# --------------------------------------------------------------------------- -# 0. Stub heavy dependencies BEFORE importing any app modules -# --------------------------------------------------------------------------- - -# Ensure backend/ is on sys.path -backend_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) -if backend_dir not in sys.path: - sys.path.insert(0, backend_dir) - -# Stub pydantic_settings so config doesn't fail -if "pydantic_settings" not in sys.modules: - pydantic_settings_mock = ModuleType("pydantic_settings") - - class _BaseSettings: - def __init__(self, **kwargs): - pass - - def __init_subclass__(cls, **kwargs): - super().__init_subclass__(**kwargs) - - pydantic_settings_mock.BaseSettings = _BaseSettings - sys.modules["pydantic_settings"] = pydantic_settings_mock - -# Stub app.config -config_mod = ModuleType("app.config") - - -class _FakeSettings: - DATABASE_URL = "sqlite:///:memory:" - SECRET_KEY = "test" - ALGORITHM = "HS256" - ACCESS_TOKEN_EXPIRE_MINUTES = 60 - REDIS_URL = "redis://localhost:6379/0" - MINIO_ENDPOINT = "localhost:9000" - MINIO_ACCESS_KEY = "test" - MINIO_SECRET_KEY = "test" - MINIO_BUCKET = "test" - MINIO_SECURE = False - MAX_RETEST_COUNT = 3 - CORS_ORIGINS = "http://localhost:3000" - SCORING_WEIGHT_TESTS = 40 - SCORING_WEIGHT_DETECTION_RULES = 20 - SCORING_WEIGHT_D3FEND = 15 - SCORING_WEIGHT_FRESHNESS = 15 - SCORING_WEIGHT_PLATFORM_DIVERSITY = 10 - REPORT_TEMPLATES_DIR = "app/templates/reports" - REPORT_OUTPUT_DIR = "/tmp/aegis_reports" - COMPANY_NAME = "Test Org" - COMPANY_LOGO_PATH = "app/templates/reports/assets/logo.png" - JIRA_ENABLED = False - JIRA_URL = "" - JIRA_USERNAME = "" - JIRA_API_TOKEN = "" - JIRA_IS_CLOUD = True - JIRA_DEFAULT_PROJECT = "" - JIRA_ISSUE_TYPE_TEST = "Task" - JIRA_ISSUE_TYPE_CAMPAIGN = "Epic" - TEMPO_ENABLED = False - TEMPO_API_TOKEN = "" - TEMPO_DEFAULT_WORK_TYPE = "Red Team" - NVD_API_KEY = "" - STALE_THRESHOLD_DAYS = 365 - - -config_mod.settings = _FakeSettings() -sys.modules["app.config"] = config_mod - -# Stub app.database so no real engine is created -db_mod = ModuleType("app.database") -db_mod.Base = type("Base", (), {"metadata": MagicMock()}) -db_mod.get_db = MagicMock() -sys.modules["app.database"] = db_mod - -# Stub taxii2client -taxii_v20 = ModuleType("taxii2client.v20") -taxii_v20.Server = MagicMock -sys.modules["taxii2client"] = ModuleType("taxii2client") -sys.modules["taxii2client.v20"] = taxii_v20 - -# Stub jose -jose_mod = ModuleType("jose") -jose_mod.JWTError = Exception -jose_mod.jwt = MagicMock() -sys.modules["jose"] = jose_mod - -# Stub boto3 -boto3_mod = ModuleType("boto3") -boto3_mod.client = MagicMock() -sys.modules["boto3"] = boto3_mod -sys.modules["botocore"] = ModuleType("botocore") -sys.modules["botocore.exceptions"] = ModuleType("botocore.exceptions") -sys.modules["botocore.exceptions"].ClientError = Exception - -# Stub apscheduler -sys.modules["apscheduler"] = ModuleType("apscheduler") -sys.modules["apscheduler.schedulers"] = ModuleType("apscheduler.schedulers") -sys.modules["apscheduler.schedulers.background"] = ModuleType("apscheduler.schedulers.background") -sys.modules["apscheduler.schedulers.background"].BackgroundScheduler = MagicMock -sys.modules["apscheduler.triggers"] = ModuleType("apscheduler.triggers") -sys.modules["apscheduler.triggers.cron"] = ModuleType("apscheduler.triggers.cron") -sys.modules["apscheduler.triggers.cron"].CronTrigger = MagicMock - -# --------------------------------------------------------------------------- -# Now we can safely import -# --------------------------------------------------------------------------- +from app.domain.exceptions import InvalidTransitionError from app.models.enums import TestState from app.services.test_workflow_service import ( - VALID_TRANSITIONS, can_transition, - transition_state, - start_execution, - submit_red_evidence, - submit_blue_evidence, - validate_as_red_lead, - validate_as_blue_lead, - check_dual_validation, reopen_test, + start_execution, + submit_blue_evidence, + submit_red_evidence, + transition_state, + validate_as_blue_lead, + validate_as_red_lead, ) -# We need the domain exceptions for assertions -from app.domain.exceptions import InvalidTransitionError - # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- diff --git a/backend/tests/test_tests.py b/backend/tests/test_tests.py index 84ad8c5..bd669c8 100644 --- a/backend/tests/test_tests.py +++ b/backend/tests/test_tests.py @@ -75,3 +75,45 @@ def test_get_test_by_id(client, auth_headers, technique): response = client.get(f"/api/v1/tests/{test_id}", headers=auth_headers) assert response.status_code == 200 assert response.json()["id"] == test_id + + +def test_start_execution_twice_returns_invalid_transition( + client, auth_headers, technique, red_tech_user +): + """Invalid workflow transition surfaces domain error JSON (FASE 0.4). + + HttpOnly login cookies take precedence over the Authorization header. + Clear cookies before each phase so Bearer tokens match the intended user. + """ + client.cookies.clear() + create_response = client.post( + "/api/v1/tests", + json={"technique_id": technique["id"], "name": "Workflow dup start"}, + headers=auth_headers, + ) + assert create_response.status_code == 201 + test_id = create_response.json()["id"] + + rl = client.post( + "/api/v1/auth/login", + data={"username": "redtech", "password": "redtech123"}, + ) + assert rl.status_code == 200 + red_headers = {"Authorization": f"Bearer {rl.json()['access_token']}"} + client.cookies.clear() + + first = client.post( + f"/api/v1/tests/{test_id}/start-execution", + headers=red_headers, + ) + assert first.status_code == 200 + + client.cookies.clear() + second = client.post( + f"/api/v1/tests/{test_id}/start-execution", + headers=red_headers, + ) + assert second.status_code == 400 + body = second.json() + assert body.get("code") == "INVALID_TRANSITION" + assert "detail" in body