361 lines
22 KiB
Python
361 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import threading
|
|
import time
|
|
from io import StringIO
|
|
|
|
from django.core.management import call_command
|
|
|
|
from agents.venture_discovery import DEFAULT_HARD_EXCLUSIONS, EVIDENCE_CEILINGS, SCORE_DEFINITIONS, SCORE_DIMENSIONS, VentureDiscoveryService
|
|
from control_plane.ventures.models import CohortIdeationMandate, CompanyProposal, EvidenceTier, NoveltyGateDecision, OpportunityTerritory, OverlapClassification, PortfolioICReview, PortfolioSaturationAnalysis, PortfolioThesis, PortfolioThesisCluster, PortfolioThesisStatus, VentureCapabilityDemand, VentureCohort, VentureCollision, VentureGenerationRejection, VentureThesisFingerprint
|
|
from graph.bootstrap import champion_venture_discovery_cohort_graph_v1
|
|
from graph.langgraph_runtime import LangGraphRuntime
|
|
from graph.models import GraphRun, GraphRunStatus
|
|
from graph.venture_cohort import venture_discovery_cohort_registry
|
|
from model_router.router import ModelProvider, ModelRequestContract, ModelResponseContract, ModelRouter
|
|
|
|
|
|
class SequenceProvider(ModelProvider):
|
|
provider_name = "sol-sequence"
|
|
|
|
def __init__(self) -> None:
|
|
self.company_calls = 0
|
|
self.research_calls = 0
|
|
|
|
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
|
|
if "Bounded public web market research" in request.prompt:
|
|
self.research_calls += 1
|
|
categories = ["competitors", "pricing", "customer_pain", "market_alternatives", "regulatory_platform_risks"]
|
|
return ModelResponseContract("luna", json.dumps({"sources": [{"url": f"https://example.com/{self.research_calls}/{cat}", "title": cat, "category": cat, "summary": cat} for cat in categories], "findings": {cat: [cat] for cat in categories}, "coverage": {cat: True for cat in categories}}), {})
|
|
self.company_calls += 1
|
|
n = self.company_calls
|
|
industries = ["Shopify", "Developer", "Legal", "Healthcare", "Real Estate", "Restaurant", "Security", "Education", "Logistics", "Finance"]
|
|
industry = industries[(n - 1) % len(industries)]
|
|
return ModelResponseContract("sol", json.dumps({"title": f"{industry} Workflow Monitor {n}", "one_line_thesis": f"Sell a recurring {industry.lower()} workflow monitor to a narrow buyer before building broad software.", "description": f"A focused {industry} workflow monitoring product with AI-assisted weekly exception reports.", "problem": f"{industry} buyers miss recurring operational exceptions that cost time or revenue.", "target_customer": f"Small {industry} operators with active revenue.", "proposed_solution": f"AI-assisted {industry} workflow monitor with prioritized exception reports and lightweight automation.", "business_model": "Recurring AI-enabled service with software automation", "pricing_hypothesis": "$99/month monitor", "acquisition_strategy": "Compliant community posts and direct referrals after approval", "validation_plan": "Collect 5 credible target-customer responses or 1 willingness-to-pay signal before build/spend.", "capital_requested": "50", "time_to_first_dollar_estimate": "3-7 days after outreach approval", "expected_margin": "80-90% gross margin", "build_complexity": "LOW", "market_evidence": [], "differentiation": "Specific recurring workflow data and AI-assisted exception monitoring", "exception_rationale": "vertical specific recurring workflow with proprietary data", "major_risks": ["Demand unproven"], "confidence": 0.65}), {})
|
|
|
|
def health(self) -> str:
|
|
return "AVAILABLE"
|
|
|
|
|
|
class RejectThenAcceptProvider(SequenceProvider):
|
|
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
|
|
if "Bounded public web market research" in request.prompt:
|
|
return super().complete(request)
|
|
time.sleep(0.02)
|
|
self.company_calls += 1
|
|
if self.company_calls == 1:
|
|
return ModelResponseContract("sol", json.dumps({"title": "RFP Copilot", "one_line_thesis": "AI RFP response drafting for SaaS teams.", "description": "Automates request for proposal answers.", "problem": "Sales teams hate RFPs.", "target_customer": "B2B SaaS sales teams", "proposed_solution": "Proposal drafting automation", "business_model": "SaaS", "pricing_hypothesis": "$99/month", "acquisition_strategy": "Content", "validation_plan": "Interview later", "capital_requested": "50", "time_to_first_dollar_estimate": "7 days", "expected_margin": "90%", "build_complexity": "LOW", "market_evidence": [], "differentiation": "AI", "major_risks": ["Excluded"], "confidence": 0.5}), {})
|
|
return super().complete(request)
|
|
|
|
|
|
class ManyRejectThenAcceptProvider(SequenceProvider):
|
|
def complete(self, request: ModelRequestContract) -> ModelResponseContract:
|
|
if "Bounded public web market research" in request.prompt:
|
|
return super().complete(request)
|
|
time.sleep(0.02)
|
|
self.company_calls += 1
|
|
if self.company_calls <= 4:
|
|
return ModelResponseContract("sol", json.dumps({"title": f"RFP Copilot {self.company_calls}", "one_line_thesis": "AI RFP response drafting for SaaS teams.", "description": "Automates request for proposal answers.", "problem": "Sales teams hate RFPs.", "target_customer": "B2B SaaS sales teams", "proposed_solution": "Proposal drafting automation", "business_model": "SaaS", "pricing_hypothesis": "$99/month", "acquisition_strategy": "Content", "validation_plan": "Interview later", "capital_requested": "50", "time_to_first_dollar_estimate": "7 days", "expected_margin": "90%", "build_complexity": "LOW", "market_evidence": [], "differentiation": "AI", "major_risks": ["Excluded"], "confidence": 0.5}), {})
|
|
return super().complete(request)
|
|
|
|
|
|
def service() -> VentureDiscoveryService:
|
|
provider = SequenceProvider()
|
|
return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True, ideation_model_hint="sol", research_model_hint="luna")
|
|
|
|
|
|
def rejecting_service() -> VentureDiscoveryService:
|
|
provider = RejectThenAcceptProvider()
|
|
return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True, ideation_model_hint="sol", research_model_hint="luna")
|
|
|
|
|
|
def many_rejecting_service() -> VentureDiscoveryService:
|
|
provider = ManyRejectThenAcceptProvider()
|
|
return VentureDiscoveryService(ModelRouter({"sol": provider, "luna": provider}), web_research_available=True, ideation_model_hint="sol", research_model_hint="luna")
|
|
|
|
|
|
def test_bounded_map_runs_concurrently_and_preserves_result_order() -> None:
|
|
svc = VentureDiscoveryService()
|
|
svc._effective_concurrency = lambda concurrency: concurrency
|
|
active = 0
|
|
observed_peak = 0
|
|
lock = threading.Lock()
|
|
|
|
def worker(value: int) -> int:
|
|
nonlocal active, observed_peak
|
|
with lock:
|
|
active += 1
|
|
observed_peak = max(observed_peak, active)
|
|
time.sleep(0.02)
|
|
with lock:
|
|
active -= 1
|
|
return value * 2
|
|
|
|
assert svc._bounded_map([1, 2, 3, 4], worker, 2) == [2, 4, 6, 8]
|
|
assert observed_peak == 2
|
|
assert svc._last_bounded_map_peak == 2
|
|
|
|
|
|
def test_evidence_tiers_probability_ceiling_and_score_orientation() -> None:
|
|
svc = service()
|
|
proposal = svc.generate_single_company(svc.create_v0_mandate())
|
|
assert proposal.evidence_tier == EvidenceTier.TIER_0_THESIS
|
|
assert svc.calibrate_probability(proposal, raw_probability=80)["evidence_adjusted_probability"] == EVIDENCE_CEILINGS[EvidenceTier.TIER_0_THESIS]
|
|
svc.conduct_market_research(proposal)
|
|
assert proposal.evidence_tier == EvidenceTier.TIER_1_PUBLIC_EVIDENCE
|
|
assert svc.calibrate_probability(proposal, raw_probability=80)["evidence_adjusted_probability"] == EVIDENCE_CEILINGS[EvidenceTier.TIER_1_PUBLIC_EVIDENCE]
|
|
diligence = svc.start_ic_diligence(proposal)
|
|
svc.generate_ic_questions(diligence)
|
|
svc.answer_questions(diligence)
|
|
svc.red_team(diligence)
|
|
svc.final_company_response(diligence)
|
|
decision = svc.score_and_decide(diligence)
|
|
assert set(decision.component_scores) == set(SCORE_DIMENSIONS)
|
|
assert "AI Leverage" in decision.component_scores
|
|
assert "Platformization Potential" in decision.component_scores
|
|
assert all("100 = highly attractive" in definition for definition in decision.score_definitions.values())
|
|
assert decision.probability_500_within_30_days <= decision.evidence_ceiling
|
|
assert decision.raw_probability_500_within_30_days >= decision.probability_500_within_30_days
|
|
|
|
|
|
def test_ai_native_platform_opportunities_receive_portfolio_preference() -> None:
|
|
svc = service()
|
|
mandate = svc.create_v0_mandate()
|
|
ai = svc.generate_single_company(mandate)
|
|
ai.title = "AI Support Knowledge Copilot"
|
|
ai.description = "AI agent platform that monitors support tickets, synthesizes answers, and builds a reusable knowledge workflow."
|
|
ai.proposed_solution = "Agent-assisted operational software with recurring automation and data accumulation."
|
|
ai.business_model = "Monthly SaaS platform plus AI-enabled service onboarding."
|
|
ai.differentiation = "Local inference and agent workflows automate most delivery with low marginal labor."
|
|
ai.save(update_fields=["title", "description", "proposed_solution", "business_model", "differentiation", "updated_at"])
|
|
generic = svc.generate_single_company(mandate)
|
|
generic.title = "Emergency Shopify Audit Fix"
|
|
generic.description = "One-time manual Shopify audit and emergency fix service."
|
|
generic.proposed_solution = "Manual one-time consulting audit."
|
|
generic.business_model = "One-time consulting service."
|
|
generic.differentiation = "Fast human review."
|
|
generic.save(update_fields=["title", "description", "proposed_solution", "business_model", "differentiation", "updated_at"])
|
|
|
|
assert svc.ai_leverage_score(ai) > svc.ai_leverage_score(generic)
|
|
assert svc.platformization_potential(ai) > svc.platformization_potential(generic)
|
|
assert svc._generic_concentration_penalty(generic) > svc._generic_concentration_penalty(ai)
|
|
|
|
|
|
def test_identity_contamination_detection_and_fingerprint_collision() -> None:
|
|
svc = service()
|
|
proposal = svc.generate_single_company(svc.create_v0_mandate())
|
|
contaminated = {"Company": "Unrelated CRM SaaS", "ICP": "enterprise banks", "Problem": "loan defaults", "Product": "risk platform"}
|
|
result = svc.validate_identity_content(proposal, contaminated)
|
|
assert result["passed"] is False
|
|
fp = svc.fingerprint_proposal(proposal)
|
|
assert isinstance(fp, VentureThesisFingerprint)
|
|
other = svc.generate_single_company(proposal.mandate)
|
|
other.title = proposal.title + " Copy"
|
|
other.target_customer = proposal.target_customer
|
|
other.problem = proposal.problem
|
|
other.proposed_solution = proposal.proposed_solution
|
|
other.business_model = proposal.business_model
|
|
other.save(update_fields=["title", "target_customer", "problem", "proposed_solution", "business_model", "updated_at"])
|
|
overlap = svc.classify_overlap(proposal, other)
|
|
assert overlap["classification"] in {OverlapClassification.NEAR_DUPLICATE, OverlapClassification.DUPLICATE, OverlapClassification.COMPETITIVE}
|
|
|
|
|
|
def test_cohort_size_ranking_top3_capability_aggregation_and_graph_lineage() -> None:
|
|
svc = service()
|
|
version = champion_venture_discovery_cohort_graph_v1()
|
|
graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"])
|
|
LangGraphRuntime(venture_discovery_cohort_registry(svc, cohort_size=10, concurrency=2)).run_until_terminal_or_paused(graph_run)
|
|
graph_run.refresh_from_db()
|
|
cohort = VentureCohort.objects.get(id=graph_run.metadata["cohort_id"])
|
|
|
|
assert graph_run.status == GraphRunStatus.COMPLETE
|
|
assert cohort.members.count() == 10
|
|
assert CompanyProposal.objects.count() == 10
|
|
assert cohort.members.filter(is_top_3=True).count() == len(cohort.portfolio_review.top_3)
|
|
assert PortfolioICReview.objects.filter(cohort=cohort).exists()
|
|
assert VentureCollision.objects.filter(cohort=cohort).count() == 45
|
|
assert VentureCapabilityDemand.objects.filter(cohort=cohort).exists()
|
|
assert cohort.metadata["real_spend"] == 0
|
|
assert cohort.metadata["real_customer_outreach"] is False
|
|
assert cohort.concurrency == 2
|
|
assert cohort.metrics["peak_concurrency"] == 1
|
|
assert cohort.metrics["proposal_generation_peak_concurrency"] == 1
|
|
assert cohort.metrics["research_peak_concurrency"] == 1
|
|
assert cohort.metrics["individual_diligence_peak_concurrency"] == 1
|
|
assert cohort.metrics["finalist_deep_research_count"] == 5
|
|
assert "ranking_changes_after_deep_research" in cohort.metrics
|
|
report = cohort.mandate.artifacts.get(artifact_type="VENTURE_DISCOVERY_COHORT_REPORT")
|
|
assert len(report.content["rankings"]) == 10
|
|
assert len(report.content["top_3"]) <= 3
|
|
assert report.content["total_spend"] == 0
|
|
assert CohortIdeationMandate.objects.filter(cohort=cohort).exists()
|
|
assert PortfolioThesisCluster.objects.filter(cohort=cohort).count() == 10
|
|
assert PortfolioSaturationAnalysis.objects.filter(cohort=cohort).exists()
|
|
assert "ideation_mandate" in report.content
|
|
assert "thesis_registry_before" in report.content
|
|
assert "thesis_registry_after" in report.content
|
|
assert "saturation_analysis" in report.content
|
|
assert "idea_diversity_metrics" in report.content
|
|
|
|
|
|
def test_v03_thesis_registry_mandate_and_territory_allocation() -> None:
|
|
svc = service()
|
|
cohort = svc.prepare_cohort(size=10)
|
|
review = svc.portfolio_thesis_review(cohort)
|
|
mandate = svc.create_ideation_mandate(cohort)
|
|
|
|
assert PortfolioThesis.objects.filter(canonical_name="AI RFP response automation for B2B SaaS", status=PortfolioThesisStatus.SATURATED).exists()
|
|
assert "AI RFP response automation for B2B SaaS" in [row["canonical_name"] for row in review["saturated_thesis_areas"]]
|
|
assert mandate.hard_exclusions == DEFAULT_HARD_EXCLUSIONS
|
|
assert mandate.opportunity_territories[:2] == [OpportunityTerritory.VERTICAL_AI_WRAPPERS, OpportunityTerritory.VERTICAL_AI_WRAPPERS]
|
|
assert mandate.opportunity_territories[-2:] == [OpportunityTerritory.OPEN_CATEGORY, OpportunityTerritory.OPEN_CATEGORY]
|
|
|
|
|
|
def test_v03_registry_seed_historical_metadata_is_json_serializable() -> None:
|
|
svc = service()
|
|
proposal = svc.generate_single_company(svc.create_v0_mandate())
|
|
proposal.title = "RFP Historical Candidate"
|
|
proposal.description = "AI RFP response drafting for SaaS teams."
|
|
proposal.proposed_solution = "RFP proposal automation."
|
|
proposal.save(update_fields=["title", "description", "proposed_solution", "updated_at"])
|
|
|
|
svc.seed_dogfood_thesis_registry()
|
|
|
|
thesis = PortfolioThesis.objects.get(canonical_name="AI RFP response automation for B2B SaaS")
|
|
assert isinstance(thesis.metadata["historical"]["best_company"], str)
|
|
|
|
|
|
def test_v03_novelty_gate_blocks_hard_exclusion_and_allows_soft_exception() -> None:
|
|
svc = service()
|
|
cohort = svc.prepare_cohort(size=2)
|
|
mandate = svc.create_ideation_mandate(cohort)
|
|
|
|
hard_payload = {"title": "RFP Copilot", "one_line_thesis": "AI RFP response drafting", "description": "Drafts proposal responses", "problem": "RFPs", "target_customer": "SaaS sales", "proposed_solution": "RFP automation", "business_model": "SaaS", "pricing_hypothesis": "$99", "acquisition_strategy": "content", "validation_plan": "test", "differentiation": "AI"}
|
|
soft_payload = {**hard_payload, "title": "Regulated Contract Workflow Monitor", "one_line_thesis": "Contract scanner for regulated vendor workflows", "description": "Contract scanner with proprietary recurring workflow data", "problem": "Regulated teams miss renewal obligations", "target_customer": "Compliance teams", "proposed_solution": "Vertical contract workflow monitor", "differentiation": "Vertical proprietary workflow data", "exception_rationale": "specific regulated workflow with recurring data"}
|
|
|
|
assert svc.novelty_gate(hard_payload, cohort, [], mandate, territory="OPEN_CATEGORY")["decision"] == NoveltyGateDecision.REGENERATE_HARD_EXCLUSION
|
|
assert svc.novelty_gate(soft_payload, cohort, [], mandate, territory="DATA_DOCUMENT_AUTOMATION")["decision"] == NoveltyGateDecision.ACCEPT
|
|
|
|
|
|
def test_v03_generation_regenerates_rejected_slots_and_preserves_size() -> None:
|
|
svc = rejecting_service()
|
|
cohort = svc.prepare_cohort(size=3)
|
|
svc.portfolio_thesis_review(cohort)
|
|
svc.create_ideation_mandate(cohort)
|
|
|
|
proposals = svc.generate_independent_proposals(cohort)
|
|
|
|
assert len(proposals) == 3
|
|
assert cohort.members.count() == 3
|
|
assert VentureGenerationRejection.objects.filter(cohort=cohort, decision=NoveltyGateDecision.REGENERATE_HARD_EXCLUSION).count() == 1
|
|
assert cohort.metrics["hard_exclusion_rejections"] == 1
|
|
assert cohort.metrics["failed_slots"] == 0
|
|
|
|
|
|
def test_v04_generation_uses_concurrency_and_cohort_attempt_pool() -> None:
|
|
svc = many_rejecting_service()
|
|
svc._effective_concurrency = lambda concurrency: concurrency
|
|
cohort = svc.prepare_cohort(size=3, concurrency=2)
|
|
cohort.scoring_policy = {**cohort.scoring_policy, "duplicate_policy": {**cohort.scoring_policy["duplicate_policy"], "cohort_attempt_budget_multiplier": 4}}
|
|
cohort.save(update_fields=["scoring_policy", "updated_at"])
|
|
svc.portfolio_thesis_review(cohort)
|
|
svc.create_ideation_mandate(cohort)
|
|
|
|
proposals = svc.generate_independent_proposals(cohort)
|
|
|
|
assert len(proposals) == 3
|
|
assert cohort.metrics["proposal_generation_peak_concurrency"] == 2
|
|
assert cohort.metrics["replacement_attempts"] > 0
|
|
assert cohort.metrics["generation_attempts"] > cohort.cohort_size
|
|
assert cohort.status == "PROPOSALS_GENERATED"
|
|
|
|
|
|
def test_v04_partial_complete_when_attempt_budget_exhausted() -> None:
|
|
svc = rejecting_service()
|
|
cohort = svc.prepare_cohort(size=3)
|
|
cohort.scoring_policy = {**cohort.scoring_policy, "duplicate_policy": {**cohort.scoring_policy["duplicate_policy"], "cohort_attempt_budget_multiplier": 1}}
|
|
cohort.save(update_fields=["scoring_policy", "updated_at"])
|
|
svc.portfolio_thesis_review(cohort)
|
|
svc.create_ideation_mandate(cohort)
|
|
|
|
proposals = svc.generate_independent_proposals(cohort)
|
|
|
|
assert len(proposals) < 3
|
|
assert cohort.status == "INSUFFICIENT_ACCEPTED_PROPOSALS"
|
|
|
|
|
|
def test_v04_semantic_duplicate_clusters_known_variants() -> None:
|
|
svc = service()
|
|
cohort = svc.prepare_cohort(size=3)
|
|
mandate = svc.create_ideation_mandate(cohort)
|
|
first = {"title": "PermitDoc Compliance Checker", "one_line_thesis": "Municipal permit compliance checker", "description": "Checks permit compliance for contractors", "problem": "Permit misses", "target_customer": "Residential contractors", "proposed_solution": "Permit compliance checker", "business_model": "SaaS", "pricing_hypothesis": "$99", "acquisition_strategy": "content", "validation_plan": "test", "differentiation": "workflow"}
|
|
second = {**first, "title": "PermitDoc Pre-check", "one_line_thesis": "Municipal permit pre-check for contractors"}
|
|
legacy = {**first, "title": "Legacy Modernization Triage Agent", "one_line_thesis": "Legacy modernization copilot triages risky code migrations", "description": "Legacy modernization triage agent", "problem": "legacy migration risk", "proposed_solution": "triage agent"}
|
|
|
|
assert svc.current_cohort_match(second, [first])["classification"] == "NEAR_DUPLICATE"
|
|
assert svc.semantic_cluster_key(first) == svc.semantic_cluster_key(second)
|
|
assert svc.semantic_cluster_key(legacy) == "legacy_modernization_triage"
|
|
assert svc.novelty_gate(second, cohort, [first, second], mandate, territory="OPEN_CATEGORY")["decision"] == NoveltyGateDecision.REGENERATE_DUPLICATE
|
|
|
|
|
|
def test_v04_research_source_quality_filters_irrelevant_pages() -> None:
|
|
svc = service()
|
|
proposal = svc.generate_single_company(svc.create_v0_mandate())
|
|
sources = [
|
|
{"url": "https://example.com/pricing", "title": "Pricing plans for workflow automation", "category": "pricing", "summary": proposal.title + " price cost subscription"},
|
|
{"url": "https://archive.org/noise", "title": "Archived unrelated page", "category": "pricing", "summary": "celebrity news"},
|
|
{"url": "https://example.com/noise", "title": "Sports scores", "category": "pricing", "summary": "football fixtures"},
|
|
]
|
|
|
|
quality = svc.filter_research_sources(proposal, sources, ["pricing"])
|
|
|
|
assert len(quality["accepted_sources"]) == 1
|
|
assert quality["accepted_sources"][0]["quality"] == "strong"
|
|
assert len(quality["rejected_sources"]) == 2
|
|
|
|
|
|
def test_run_venture_cohort_management_command_outputs_summary(monkeypatch) -> None:
|
|
import control_plane.ventures.management.commands.run_venture_cohort as command_module
|
|
|
|
provider = SequenceProvider()
|
|
monkeypatch.setattr(command_module, "providers_from_resources", lambda: {"qwen": provider})
|
|
stdout = StringIO()
|
|
|
|
call_command("run_venture_cohort", "--size", "3", "--concurrency", "2", "--qwen-only", "--no-web-research", stdout=stdout)
|
|
|
|
summary = json.loads(stdout.getvalue())
|
|
cohort = VentureCohort.objects.get(cohort_id=summary["cohort_id"])
|
|
|
|
assert summary["status"] == GraphRunStatus.COMPLETE
|
|
assert summary["members"] == 3
|
|
assert summary["concurrency"] == 2
|
|
assert summary["fallback_count"] == 0
|
|
assert summary["generation_sources"] == ["qwen"]
|
|
assert len(summary["top_3"]) <= 3
|
|
assert cohort.members.count() == 3
|
|
|
|
|
|
def test_export_venture_cohort_management_command_outputs_all_companies(tmp_path) -> None:
|
|
svc = service()
|
|
version = champion_venture_discovery_cohort_graph_v1()
|
|
graph_run = GraphRun.objects.create(execution_graph_version=version, current_node=version.graph_spec["entry"])
|
|
LangGraphRuntime(venture_discovery_cohort_registry(svc, cohort_size=3, concurrency=1)).run_until_terminal_or_paused(graph_run)
|
|
cohort = VentureCohort.objects.get(id=graph_run.metadata["cohort_id"])
|
|
markdown_path = tmp_path / "cohort.md"
|
|
json_path = tmp_path / "cohort.json"
|
|
|
|
call_command("export_venture_cohort", "--cohort", cohort.cohort_id, "--output", str(markdown_path))
|
|
call_command("export_venture_cohort", "--cohort", cohort.cohort_id, "--format", "json", "--output", str(json_path))
|
|
|
|
markdown = markdown_path.read_text(encoding="utf-8")
|
|
payload = json.loads(json_path.read_text(encoding="utf-8"))
|
|
|
|
assert "## All Company Details" in markdown
|
|
assert markdown.count("### Rank") == 3
|
|
assert "Validation condition:" in markdown
|
|
assert "Autonomous operability:" in markdown
|
|
assert payload["cohort"]["cohort_id"] == cohort.cohort_id
|
|
assert len(payload["companies"]) == 3
|
|
assert payload["companies"][0]["decision"]["decision"]
|
|
assert payload["companies"][0]["autonomous_assessment"]["venture_track"]
|