| import os |
| import sys |
|
|
| import pytest |
| from fastapi.testclient import TestClient |
|
|
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) |
|
|
|
|
| @pytest.fixture(autouse=True) |
| def temp_env(monkeypatch, tmp_path): |
| db_path = str(tmp_path / "test_vibe.db") |
| model_path = str(tmp_path / "test_model.pkl") |
| monkeypatch.setattr("app.config.settings.DB_PATH", db_path) |
| monkeypatch.setattr("app.config.settings.MODEL_PATH", model_path) |
|
|
|
|
| @pytest.fixture |
| def client(temp_env): |
| from app.main import app |
| from app.api.routes import init_engines |
| from app.data.database import init_db |
|
|
| init_db() |
| init_engines() |
|
|
| return TestClient(app) |
|
|
|
|
| def test_health(client): |
| resp = client.get("/health") |
| assert resp.status_code == 200 |
| assert resp.json()["status"] == "ok" |
|
|
|
|
| def test_scan_file(client): |
| import random |
| rng = random.Random(42) |
| data = bytes(rng.randint(0, 255) for _ in range(4096)) |
|
|
| resp = client.post("/scan", files={"file": ("test.bin", data, "application/octet-stream")}) |
| assert resp.status_code == 200 |
| body = resp.json() |
| assert "sha256" in body |
| assert "vibe_id" in body |
| assert "final_verdict" in body |
| assert "pipeline_stages_run" in body |
| assert body["is_new_vibe"] is True |
|
|
|
|
| def test_scan_duplicate(client): |
| import random |
| rng = random.Random(99) |
| data = bytes(rng.randint(0, 255) for _ in range(4096)) |
|
|
| client.post("/scan", files={"file": ("test.bin", data, "application/octet-stream")}) |
| resp = client.post("/scan", files={"file": ("test.bin", data, "application/octet-stream")}) |
| assert resp.status_code == 200 |
| body = resp.json() |
| assert body["is_new_vibe"] is False |
| assert body["final_verdict"] in ("KNOWN_MUTATION", "VIBE_MUTATION") |
|
|
|
|
| def test_scan_empty_file(client): |
| resp = client.post("/scan", files={"file": ("empty.bin", b"", "application/octet-stream")}) |
| assert resp.status_code == 400 |
|
|
|
|
| def test_stats(client): |
| resp = client.get("/stats") |
| assert resp.status_code == 200 |
| body = resp.json() |
| assert "total_vibes" in body |
| assert "total_samples" in body |
|
|
|
|
| def test_vibe_not_found(client): |
| resp = client.get("/vibes/nonexistent-id") |
| assert resp.status_code == 404 |
|
|
|
|
| def test_scan_then_get_vibe(client): |
| import random |
| rng = random.Random(77) |
| data = bytes(rng.randint(0, 255) for _ in range(4096)) |
|
|
| scan_resp = client.post("/scan", files={"file": ("test.bin", data, "application/octet-stream")}) |
| vibe_id = scan_resp.json()["vibe_id"] |
|
|
| resp = client.get(f"/vibes/{vibe_id}") |
| assert resp.status_code == 200 |
| body = resp.json() |
| assert body["vibe_id"] == vibe_id |
| assert len(body["samples"]) >= 1 |
|
|