| import json |
| from pathlib import Path |
| import pytest |
| from nato1000_runtime.config import RuntimeConfig, IngestionConfig |
| from nato1000_runtime.ingestion import DataIngestionPipeline, IngestionError |
| from nato1000_runtime.models import MODEL_REGISTRY |
| from nato1000_runtime.observability import AuditLedger |
| from nato1000_runtime.runtime import NATO1000Runtime, RuntimeFailure |
|
|
|
|
| def test_all_models_run(tmp_path, monkeypatch): |
| monkeypatch.chdir(tmp_path) |
| for model_id in MODEL_REGISTRY: |
| result = NATO1000Runtime(RuntimeConfig(model_id=model_id)).run("test prompt") |
| assert result.model_id == model_id |
| assert result.text |
|
|
|
|
| def test_config_rejects_invalid_temperature(): |
| with pytest.raises(ValueError): |
| RuntimeConfig(temperature=4).validate() |
|
|
|
|
| def test_ingestion_redacts_and_deduplicates(tmp_path, monkeypatch): |
| monkeypatch.chdir(tmp_path) |
| source = tmp_path / "sample.txt" |
| source.write_text("api_key=SECRET123\nhello") |
| pipeline = DataIngestionPipeline(IngestionConfig()) |
| rows = pipeline.ingest_file(source) |
| assert len(rows) == 1 |
| assert "SECRET123" not in rows[0].content |
| assert "[REDACTED]" in rows[0].content |
| assert pipeline.ingest_file(source) == [] |
|
|
|
|
| def test_ingestion_quarantines_oversized_file(tmp_path, monkeypatch): |
| monkeypatch.chdir(tmp_path) |
| source = tmp_path / "large.txt" |
| source.write_text("x" * 2048) |
| with pytest.raises(IngestionError): |
| DataIngestionPipeline(IngestionConfig(max_bytes=1024)).ingest_file(source) |
| assert (tmp_path / "data/quarantine/large.txt").exists() |
|
|
|
|
| def test_audit_chain_detects_tampering(tmp_path): |
| ledger = AuditLedger(str(tmp_path / "audit.jsonl")) |
| ledger.record("test", "1", "unit", "success", "ok") |
| assert ledger.verify() |
| path = tmp_path / "audit.jsonl" |
| data = json.loads(path.read_text()) |
| data["message"] = "tampered" |
| path.write_text(json.dumps(data) + "\n") |
| assert not ledger.verify() |
|
|
|
|
| def test_fail_closed_on_empty_prompt(tmp_path, monkeypatch): |
| monkeypatch.chdir(tmp_path) |
| with pytest.raises(RuntimeFailure): |
| NATO1000Runtime(RuntimeConfig()).run(" ") |
|
|
