Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86ec451f74
|
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,119 @@
|
||||
"""Offline unit tests: no network, no cloud or LLM credentials, no provider SDKs. They cover the pieces that read
|
||||
configuration and on-disk data and the parser for LLM replies."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
import aestimare # noqa: E402
|
||||
from analyzers import llm_analyzer # noqa: E402
|
||||
|
||||
EXAMPLE = ROOT / "config" / "accounts.example.yaml"
|
||||
|
||||
|
||||
class LoadConfigTest(unittest.TestCase):
|
||||
def test_example_config_has_the_sections_the_tool_reads(self):
|
||||
config = aestimare.load_config(str(EXAMPLE))
|
||||
self.assertIn("assessment", config)
|
||||
self.assertTrue(config["aws_accounts"])
|
||||
for account in config["aws_accounts"]:
|
||||
for key in ("name", "profile", "account_id", "regions", "enabled"):
|
||||
self.assertIn(key, account)
|
||||
self.assertRegex(str(account["account_id"]), r"^\d{12}$")
|
||||
self.assertTrue(config["github"]["organizations"])
|
||||
|
||||
def test_missing_config_falls_back_to_the_example_next_to_it(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
config_dir = Path(tmp)
|
||||
(config_dir / "accounts.example.yaml").write_text("assessment:\n name: fallback\n", encoding="utf-8")
|
||||
with mock.patch("builtins.print"):
|
||||
config = aestimare.load_config(str(config_dir / "accounts.yaml"))
|
||||
self.assertEqual(config["assessment"]["name"], "fallback")
|
||||
|
||||
def test_missing_config_without_an_example_is_an_error(self):
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
aestimare.load_config(str(Path(tmp) / "accounts.yaml"))
|
||||
|
||||
|
||||
class DataFilesTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.config = {"data_dir": self.tmp.name}
|
||||
|
||||
def test_saved_analysis_is_loaded_back_under_its_key(self):
|
||||
aestimare.save_analysis("security_analysis.json", {"findings": [1, 2]}, self.config)
|
||||
aestimare.save_analysis("cost_analysis.json", {"when": __import__("datetime").datetime(2026, 1, 2)}, self.config)
|
||||
loaded = aestimare.load_existing_analysis(self.config)
|
||||
self.assertEqual(loaded["security"], {"findings": [1, 2]})
|
||||
self.assertEqual(loaded["cost"], {"when": "2026-01-02 00:00:00"})
|
||||
self.assertNotIn("infrastructure", loaded)
|
||||
|
||||
def test_no_data_directory_loads_as_empty(self):
|
||||
missing = {"data_dir": str(Path(self.tmp.name) / "absent")}
|
||||
self.assertEqual(aestimare.load_existing_analysis(missing), {})
|
||||
self.assertEqual(aestimare.load_existing_data(missing), ({}, {}))
|
||||
|
||||
def test_collected_data_is_keyed_by_account_and_organisation(self):
|
||||
base = Path(self.tmp.name)
|
||||
for kind, name, payload in (("aws", "prod", {"a": 1}), ("github", "org", {"b": 2})):
|
||||
folder = base / kind / name
|
||||
folder.mkdir(parents=True)
|
||||
(folder / "full_data.json").write_text(json.dumps(payload), encoding="utf-8")
|
||||
(base / "aws" / "empty").mkdir()
|
||||
(base / "aws" / "stray.txt").write_text("x", encoding="utf-8")
|
||||
self.assertEqual(aestimare.load_existing_data(self.config), ({"prod": {"a": 1}}, {"org": {"b": 2}}))
|
||||
|
||||
|
||||
class ParseJsonResponseTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
with mock.patch.dict(os.environ, {"ANTHROPIC_API_KEY": "unused-test-key"}):
|
||||
self.analyzer = llm_analyzer.LLMAnalyzer({"llm": {"provider": "anthropic"}})
|
||||
|
||||
def test_fenced_json_block_is_extracted_from_prose(self):
|
||||
self.assertEqual(self.analyzer._parse_json_response('Here:\n```json\n{"score": 7}\n```\nDone'), {"score": 7})
|
||||
|
||||
def test_bare_json_is_parsed(self):
|
||||
self.assertEqual(self.analyzer._parse_json_response('{"score": 3}'), {"score": 3})
|
||||
|
||||
def test_a_broken_fence_falls_back_to_the_whole_reply_then_to_raw(self):
|
||||
broken = '```json\n{"score": }\n```'
|
||||
with self.assertLogs(llm_analyzer.logger, level="WARNING"):
|
||||
self.assertEqual(self.analyzer._parse_json_response(broken), {"raw_response": broken})
|
||||
|
||||
def test_unparseable_text_is_kept_raw_and_logged(self):
|
||||
with self.assertLogs(llm_analyzer.logger, level="WARNING"):
|
||||
self.assertEqual(self.analyzer._parse_json_response("no json"), {"raw_response": "no json"})
|
||||
|
||||
|
||||
class ProviderSelectionTest(unittest.TestCase):
|
||||
def test_unknown_provider_is_rejected(self):
|
||||
with self.assertRaisesRegex(ValueError, "Unknown LLM provider"):
|
||||
llm_analyzer.LLMAnalyzer({"llm": {"provider": "nope"}})
|
||||
|
||||
def test_anthropic_without_a_key_is_rejected(self):
|
||||
with mock.patch.dict(os.environ, {}, clear=True):
|
||||
with self.assertRaisesRegex(ValueError, "ANTHROPIC_API_KEY"):
|
||||
llm_analyzer.LLMAnalyzer({"llm": {"provider": "anthropic"}})
|
||||
|
||||
def test_local_provider_needs_no_credentials_and_takes_its_endpoint_from_config(self):
|
||||
analyzer = llm_analyzer.LLMAnalyzer({"llm": {"provider": "local", "base_url": "http://h:1", "model": "m"}})
|
||||
self.assertEqual((analyzer.provider.base_url, analyzer.provider.model), ("http://h:1", "m"))
|
||||
|
||||
def test_missing_prompt_template_is_an_error(self):
|
||||
with mock.patch.dict(os.environ, {"ANTHROPIC_API_KEY": "unused-test-key"}):
|
||||
analyzer = llm_analyzer.LLMAnalyzer({"llm": {}})
|
||||
with self.assertRaises(FileNotFoundError):
|
||||
analyzer._load_prompt("does-not-exist.md")
|
||||
self.assertTrue(analyzer._load_prompt("security_assessment.md"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user