Spaces:
Running
Running
| import pytest | |
| class TestGraphAnalyticsOffline: | |
| """Test GraphAnalytics with in-memory sample data (no Neo4j required).""" | |
| SAMPLE_NODES = [ | |
| ("P001", {"name": "Politician A", "type": "Politician"}), | |
| ("P002", {"name": "Politician B", "type": "Politician"}), | |
| ("C001", {"name": "Company X", "type": "Company"}), | |
| ("C002", {"name": "Company Y", "type": "Company"}), | |
| ("C003", {"name": "Company Z", "type": "Company"}), | |
| ("CT01", {"name": "Contract 1", "type": "Contract"}), | |
| ("CT02", {"name": "Contract 2", "type": "Contract"}), | |
| ("M001", {"name": "Ministry A", "type": "Ministry"}), | |
| ] | |
| SAMPLE_EDGES = [ | |
| ("P001", "C001", {"rel": "DIRECTOR_OF"}), | |
| ("P001", "C002", {"rel": "DIRECTOR_OF"}), | |
| ("P002", "C003", {"rel": "DIRECTOR_OF"}), | |
| ("C001", "CT01", {"rel": "WON_CONTRACT"}), | |
| ("C002", "CT01", {"rel": "WON_CONTRACT"}), | |
| ("C003", "CT02", {"rel": "WON_CONTRACT"}), | |
| ("M001", "CT01", {"rel": "AWARDED_BY"}), | |
| ("M001", "CT02", {"rel": "AWARDED_BY"}), | |
| ("P001", "P002", {"rel": "MEMBER_OF"}), | |
| ] | |
| def analytics(self): | |
| from ai.graph_analytics import GraphAnalytics | |
| return GraphAnalytics(driver=None) | |
| def test_betweenness_returns_list(self, analytics): | |
| results = analytics.compute_betweenness_centrality( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| assert isinstance(results, list) | |
| assert len(results) > 0 | |
| def test_betweenness_has_required_keys(self, analytics): | |
| results = analytics.compute_betweenness_centrality( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| required = {"entity_id", "name", "type", "betweenness_centrality"} | |
| for r in results: | |
| assert required.issubset(r.keys()), f"Missing keys in {r}" | |
| def test_betweenness_scores_between_0_and_1(self, analytics): | |
| results = analytics.compute_betweenness_centrality( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| for r in results: | |
| assert 0.0 <= r["betweenness_centrality"] <= 1.0 | |
| def test_betweenness_too_few_nodes_returns_empty(self, analytics): | |
| tiny_nodes = [("A", {}), ("B", {})] | |
| tiny_edges = [("A", "B", {})] | |
| result = analytics.compute_betweenness_centrality(tiny_nodes, tiny_edges) | |
| assert result == [] | |
| def test_pagerank_returns_list(self, analytics): | |
| results = analytics.compute_pagerank( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| assert isinstance(results, list) | |
| assert len(results) > 0 | |
| def test_pagerank_scores_positive(self, analytics): | |
| results = analytics.compute_pagerank( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| for r in results: | |
| assert r["pagerank"] > 0 | |
| def test_pagerank_sum_approximately_one(self, analytics): | |
| results = analytics.compute_pagerank( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| total = sum(r["pagerank"] for r in results) | |
| assert abs(total - 1.0) < 0.01, f"PageRank sum={total} (expected ~1.0)" | |
| def test_communities_returns_list(self, analytics): | |
| results = analytics.detect_communities( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| assert isinstance(results, list) | |
| assert len(results) > 0 | |
| def test_communities_have_members(self, analytics): | |
| results = analytics.detect_communities( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| for c in results: | |
| assert "members" in c | |
| assert len(c["members"]) >= 2 | |
| def test_communities_have_interpretation(self, analytics): | |
| results = analytics.detect_communities( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| for c in results: | |
| assert "interpretation" in c | |
| assert len(c["interpretation"]) > 0 | |
| def test_too_few_nodes_communities_empty(self, analytics): | |
| result = analytics.detect_communities( | |
| [("A", {}), ("B", {})], | |
| [("A", "B", {})] | |
| ) | |
| assert result == [] | |
| def test_run_full_analysis_no_driver(self, analytics): | |
| """With driver=None, _fetch_graph_from_neo4j returns empty.""" | |
| result = analytics.run_full_analysis() | |
| assert result.get("status") == "no_data" or "analyzed_at" in result | |
| def test_ministry_is_highest_betweenness(self, analytics): | |
| """Ministry A bridges both contracts -- should have high betweenness.""" | |
| results = analytics.compute_betweenness_centrality( | |
| self.SAMPLE_NODES, self.SAMPLE_EDGES | |
| ) | |
| top_name = results[0]["name"] if results else "" | |
| # Ministry or Politician A should be top broker | |
| assert top_name in ("Ministry A", "Politician A"), ( | |
| f"Expected Ministry A or Politician A at top, got {top_name}" | |
| ) | |
| class TestTimelineHelpers: | |
| """Test timeline helper functions without Neo4j.""" | |
| def test_label_to_category_contract(self): | |
| from api.routes.timeline import _label_to_category | |
| assert _label_to_category("Contract") == "contract" | |
| assert _label_to_category("Tender") == "contract" | |
| def test_label_to_category_legal(self): | |
| from api.routes.timeline import _label_to_category | |
| assert _label_to_category("CourtCase") == "legal" | |
| assert _label_to_category("InsolvencyOrder") == "legal" | |
| def test_label_to_category_unknown(self): | |
| from api.routes.timeline import _label_to_category | |
| assert _label_to_category("SomethingNew") == "other" | |
| def test_extract_date_order_date_preferred(self): | |
| from api.routes.timeline import _extract_date | |
| props = {"order_date": "2023-04-01", "scraped_at": "2024-01-01"} | |
| assert _extract_date(props) == "2023-04-01" | |
| def test_extract_date_fallback_scraped_at(self): | |
| from api.routes.timeline import _extract_date | |
| props = {"scraped_at": "2024-01-15", "source": "cag"} | |
| assert _extract_date(props) == "2024-01-15" | |
| def test_extract_date_none_when_no_date(self): | |
| from api.routes.timeline import _extract_date | |
| assert _extract_date({"name": "test", "source": "mca"}) is None | |