test: stabilize Phase 0 API and workflow tests [FASE-0.4]
Assert INVALID_TRANSITION JSON code on duplicate start, remove sys.modules stubs from T-106 tests, and complete boto3 stubs in integration tests.
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user