[{'Text': '"""Tests for GBS tools (get_status, add_statistic, create_source).\n\nThese mock mongodb_utils to test validation, formatting, and error paths\nwithout requiring a real MongoDB connection.\n"""\n\nimport json\nfrom unittest.mock import patch\n\nimport pytest\n\nfrom bson import ObjectId\n\n\n@pytest.fixture\ndef gbs_env():\n    """Ensure PT_RESEARCH_DATABASE_URI looks set so tools register."""\n    with patch.dict("os.environ", {"PT_RESEARCH_DATABASE_URI": "mongodb://localhost:27017/t"}, clear=False):\n        import importlib\n        import point_topic_mcp.tools.gbs_tools as gbs\n        importlib.reload(gbs)\n        yield gbs\n\n\ndef test_get_status_country_level(gbs_env):\n    """Country-level: returns coverage rollup with operator lists."""\n    gbs = gbs_env\n    op_id_a = ObjectId()\n    op_id_b = ObjectId()\n    mock_docs = [\n        {"_id": {"$oid": str(op_id_a)}, "name": "Alpha", "techs": ["FTTP"], "statCount": 5, "stats": [{"state": 3}] * 5},\n        {"_id": {"$oid": str(op_id_b)}, "name": "Beta", "techs": ["DSL"], "statCount": 0, "stats": []},\n    ]\n    mock_return = json.dumps({"cp": "2025Q1", "tp": "2025Q1", "docs": mock_docs})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.get_status(country="UK")\n        assert "GBS STATUS: UK" in result\n        assert "period 2025Q1" in result\n        assert "2 total" in result\n        assert "1 with data" in result\n        assert "1 without data" in result\n        assert "Alpha" in result\n        assert "50.0%" in result\n\n\ndef test_get_status_operator_not_found(gbs_env):\n    """Operator-level: returns error when operator doesn\'t exist."""\n    gbs = gbs_env\n    mock_return = json.dumps({"cp": "2025Q1", "tp": "2025Q1", "docs": []})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.get_status(country="UK", operator="NoSuchOp")\n        assert "not found" in result\n\n\ndef test_get_status_operator_level(gbs_env):\n    """Operator-level: returns records, gaps, state breakdown, admin URL."""\n    gbs = gbs_env\n    op_id = ObjectId()\n    mock_stats = [\n        {"type": "broadband", "tech": "FTTP", "channel": "Infrastructure",\n         "domain": "Residential", "subscribers": 100000, "state": 1},\n    ]\n    mock_docs = [{\n        "_id": {"$oid": str(op_id)}, "name": "TestOp", "techs": ["FTTP", "FTTC"],\n        "statCount": 1, "stats": mock_stats,\n    }]\n    mock_return = json.dumps({"cp": "2025Q1", "tp": "2025Q1", "docs": mock_docs})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.get_status(country="UK", operator="TestOp")\n        assert "GBS STATUS: TestOp (UK)" in result\n        assert str(op_id) in result\n        assert "Records:   1" in result\n        assert "Filled: 1 combos" in result\n        assert "pending: 1" in result\n        assert "100,000" in result\n        assert "Admin:" in result\n\n\ndef test_get_status_operator_period(gbs_env):\n    """Operator-level: explicit period works."""\n    gbs = gbs_env\n    op_id = ObjectId()\n    mock_docs = [{\n        "_id": {"$oid": str(op_id)}, "name": "TestOp", "techs": ["FTTP"],\n        "statCount": 0, "stats": [],\n    }]\n    mock_return = json.dumps({"cp": "2025Q3", "tp": "2025Q1", "docs": mock_docs})\n\n    with patch("point_topic_mcp.core.mongodb_utils._run_mongosh", return_value=mock_return):\n        result = gbs.get_status(country="UK", operator="TestOp", period="2025Q1")\n        assert "2025Q1" in result\n\n\ndef test_add_statistic_validation(gbs_env):\n    """add_statistic returns errors for invalid inputs."""\n    gbs = gbs_env\n    assert "Invalid" in gbs.add_statistic(type="invalid", operator_id="000000000000000000000001", period="2025Q1", tech="FTTP", channel="Wrong", domain="Residential", subscribers=1000)\n    assert "Invalid operator_id" in gbs.add_statistic(type="broadband", operator_id="bad", period="2025Q1", tech="FTTP", channel="Infrastructure", domain="Residential", subscribers=1000)\n    assert "Invalid period" in gbs.add_statistic(type="broadband", operator_id="000000000000000000000001", period="bad", tech="FTTP", channel="Infrastructure", domain="Residential", subscribers=1000)\n    assert "Invalid subscribers" in gbs.add_statistic(type="broadband", operator_id="000000000000000000000001", period="2025Q1", tech="FTTP", channel="Infrastructure", domain="Residential", subscribers=-1)\n\n\ndef test_add_statistic_success(gbs_env):\n    """add_statistic inserts a pending record with admin URL."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    mock_response = json.dumps({"ok": True, "insertedId": op_id, "operator": "TestOp", "country": "UK", "tech": "FTTP"})\n\n    with patch("point_topic_mcp.core.mongodb_utils.insert_statistic_safe", return_value=mock_response):\n        result = gbs.add_statistic(type="broadband", operator_id=op_id, period="2025Q3", tech="FTTP", channel="Infrastructure", domain="Residential", subscribers=50000, created_by="test-agent")\n        assert "Created statistic" in result\n        assert op_id in result\n        assert "TestOp" in result\n        assert "50,000" in result\n\n\ndef test_add_statistic_duplicate_rejected(gbs_env):\n    """add_statistic rejects duplicate (operator, period, type, tech, channel, domain)."""\n    gbs = gbs_env\n    existing_id = str(ObjectId())\n    mock_response = json.dumps({"ok": False, "error": f"Duplicate exists (id={existing_id})"})\n\n    with patch("point_topic_mcp.core.mongodb_utils.insert_statistic_safe", return_value=mock_response):\n        result = gbs.add_statistic(type="broadband", operator_id=str(ObjectId()), period="2025Q3", tech="FTTP", channel="Infrastructure", domain="Residential", subscribers=50000)\n        assert "Duplicate" in result\n        assert existing_id in result\n\n\ndef test_create_source_validation(gbs_env):\n    """create_source returns errors for invalid inputs."""\n    gbs = gbs_env\n    oid = "000000000000000000000001"\n    assert "Invalid operator_id" in gbs.create_source(operator_id="bad", year=2025, quarter=1, type="report")\n    assert "Invalid quarter" in gbs.create_source(operator_id=oid, year=2025, quarter=5, type="report")\n    assert "At least one" in gbs.create_source(operator_id=oid, year=2025, quarter=1, type="report", url="")\n\n\ndef test_create_source_success(gbs_env):\n    """create_source inserts a source record."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    source_id = str(ObjectId())\n\n    with (\n        patch("point_topic_mcp.core.mongodb_utils.find_one_operator_by_id", return_value={"_id": {"$oid": op_id}, "name": "TestOp"}),\n        patch("point_topic_mcp.core.mongodb_utils.find_sources", return_value=[]),\n        patch("point_topic_mcp.core.mongodb_utils.insert_one_source", return_value=source_id),\n    ):\n        result = gbs.create_source(operator_id=op_id, year=2025, quarter=3, type="report", url="https://example.com/r")\n        assert "Created source" in result\n        assert source_id in result\n\n\ndef test_create_source_duplicate_rejected(gbs_env):\n    """create_source rejects exact duplicates."""\n    gbs = gbs_env\n    op_id = str(ObjectId())\n    existing_id = str(ObjectId())\n    mock_existing = [{"_id": {"$oid": existing_id}, "type": "report", "url": "https://example.com/r", "fileUrl": None}]\n\n    with (\n        patch("point_topic_mcp.core.mongodb_utils.find_one_operator_by_id", return_value={"_id": {"$oid": op_id}, "name": "TestOp"}),\n        patch("point_topic_mcp.core.mongodb_utils.find_sources", return_value=mock_existing),\n    ):\n        result = gbs.create_source(operator_id=op_id, year=2025, quarter=3, type="report", url="https://example.com/r")\n        assert "Duplicate" in result\n        assert existing_id in result\n\n\ndef test_discovery_uses_prefix(gbs_env):\n    """Tool auto-discovery prefixes module name."""\n    import point_topic_mcp.tools as tools\n    all_tools = tools.get_all_tools()\n    names = [n for n, _ in all_tools]\n    assert "gbs_tools_get_status" in names\n    assert "gbs_tools_add_statistic" in names\n    assert "gbs_tools_create_source" in names\n    for old in ["gbs_tools_list_operators", "gbs_tools_get_gbs_status", "gbs_tools_gbs_get_status", "gbs_tools_gbs_add_statistic", "gbs_tools_gbs_create_source"]:\n        assert old not in names, f"Old name \'{old}\' should be gone"\n'}]