@prakhar-gitlab / ai-investment-platform
Code Commits Issues Pulls Sessions CI/CD
main
main
py 116 lines 4.89 KB
Raw
1 from datetime import datetime, timezone
2 from uuid import uuid4
3
4 import pytest
5
6 from app.models import CompanyResearchProfile, SourceClassification
7 from app.source_discovery import CandidateSearchResult, SearchDateWindow, SearchDiscoveryService, _candidate_has_positive_issuer_evidence
8
9
10 class _StaticSearchProvider:
11 provider_name = "test-search"
12
13 def __init__(self, candidates: list[CandidateSearchResult]) -> None:
14 self.candidates = candidates
15
16 async def discover(self, company, category, date_window: SearchDateWindow):
17 return self.candidates
18
19
20 def _federal_bank_profile() -> CompanyResearchProfile:
21 return CompanyResearchProfile(
22 instrument_id=uuid4(), company_id=uuid4(), company_name="FEDERAL BANK LTD",
23 aliases=["FEDERAL BANK"], provider_instrument_ids={"NSE": "FEDERALBNK"},
24 isin="INE171A01029", ticker="FEDBAN", exchange="NSE", mic="XNSE",
25 country="IN", currency="INR", known_domains=["federalbank.co.in"],
26 )
27
28
29 def _candidate(url: str, title: str = "", snippet: str = "") -> CandidateSearchResult:
30 return CandidateSearchResult(
31 title=title, url=url, snippet=snippet, discovered_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
32 provider="test-search", query_id="FINANCIAL_RESULTS:1", query='"FEDERAL BANK LTD" results',
33 category="FINANCIAL_RESULTS",
34 )
35
36
37 @pytest.mark.asyncio
38 async def test_generic_bse_result_naming_other_issuer_is_rejected_before_discovery_result() -> None:
39 profile = _federal_bank_profile()
40 search = SearchDiscoveryService(_StaticSearchProvider([
41 _candidate("https://www.bseindia.com/xml-data/corpfiling/AttachLive/TAMILNADU.pdf",
42 "TAMILNADU PETROPRODUCTS LTD financial results", "Quarterly result"),
43 ]))
44
45 assert await search.discover(profile, {"FINANCIAL_RESULTS"}, set()) == []
46 assert search.last_stats.rejected_reasons == {"COMPANY_RELEVANCE_FAILED": 1}
47 assert search.rejected_candidates[0].reason == "COMPANY_RELEVANCE_FAILED"
48
49
50 @pytest.mark.asyncio
51 async def test_trusted_isin_accepts_generic_exchange_candidate() -> None:
52 profile = _federal_bank_profile()
53 search = SearchDiscoveryService(_StaticSearchProvider([
54 _candidate("https://www.bseindia.com/xml-data/corpfiling/AttachLive/federal.pdf",
55 "Financial results", "INE171A01029 quarterly financial results"),
56 ]))
57
58 results = await search.discover(profile, {"FINANCIAL_RESULTS"}, set())
59
60 assert len(results) == 1
61 assert results[0].source.source_classification == SourceClassification.EXCHANGE
62
63
64 @pytest.mark.asyncio
65 async def test_trusted_provider_symbol_as_real_token_accepts_generic_exchange_candidate() -> None:
66 profile = _federal_bank_profile()
67 search = SearchDiscoveryService(_StaticSearchProvider([
68 _candidate("https://www.nseindia.com/corporates/federal.pdf", "FEDERALBNK financial results", "Quarterly result"),
69 ]))
70
71 assert len(await search.discover(profile, {"FINANCIAL_RESULTS"}, set())) == 1
72
73
74 @pytest.mark.asyncio
75 async def test_known_company_domain_remains_accepted_without_metadata_name() -> None:
76 profile = _federal_bank_profile()
77 search = SearchDiscoveryService(_StaticSearchProvider([
78 _candidate("https://www.federalbank.co.in/investors/results.pdf", "Investor document"),
79 ]))
80
81 results = await search.discover(profile, {"FINANCIAL_RESULTS"}, set())
82
83 assert len(results) == 1
84 assert results[0].source.source_classification == SourceClassification.OFFICIAL_COMPANY
85
86
87 def test_exchange_domain_alone_is_not_positive_issuer_evidence() -> None:
88 profile = _federal_bank_profile()
89 candidate = _candidate("https://www.bseindia.com/xml-data/corpfiling/AttachLive/filing.pdf", "Financial results")
90
91 assert _candidate_has_positive_issuer_evidence(profile, f"{candidate.title} {candidate.snippet} {candidate.url}") is False
92
93
94 @pytest.mark.asyncio
95 async def test_untrusted_local_symbol_cannot_override_verified_provider_identity() -> None:
96 profile = _federal_bank_profile()
97 search = SearchDiscoveryService(_StaticSearchProvider([
98 _candidate("https://www.nseindia.com/corporates/other.pdf",
99 "FEDBAN TAMILNADU PETROPRODUCTS LTD financial results", "Quarterly result"),
100 ]))
101
102 assert await search.discover(profile, {"FINANCIAL_RESULTS"}, set()) == []
103 assert search.last_stats.rejected_reasons == {"COMPANY_RELEVANCE_FAILED": 1}
104
105
106 @pytest.mark.asyncio
107 async def test_unsafe_and_duplicate_rejections_remain_unchanged() -> None:
108 profile = _federal_bank_profile()
109 duplicate = "https://www.federalbank.co.in/investors/results.pdf"
110 search = SearchDiscoveryService(_StaticSearchProvider([
111 _candidate("http://127.0.0.1/private", "FEDERAL BANK LTD results"),
112 _candidate(duplicate, "Investor document"),
113 ]))
114
115 assert await search.discover(profile, {"FINANCIAL_RESULTS"}, {duplicate}) == []
116 assert search.last_stats.rejected_reasons == {"DOMAIN_VALIDATION_FAILED": 1, "DUPLICATE": 1}