You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
189 lines
6.0 KiB
Python
189 lines
6.0 KiB
Python
"""
|
|
Test data catalog endpoint
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
def test_get_catalog(client: TestClient):
|
|
"""Test data catalog endpoint"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
# Check response structure
|
|
assert "categories" in data
|
|
assert "last_updated" in data
|
|
|
|
# Check categories structure
|
|
categories = data["categories"]
|
|
assert isinstance(categories, dict)
|
|
|
|
# Check expected categories exist
|
|
expected_categories = [
|
|
"Company Information",
|
|
"Income Statement",
|
|
"Balance Sheet",
|
|
"Cash Flow Statement",
|
|
"Valuation Ratios",
|
|
"Profitability Metrics",
|
|
"Growth Metrics",
|
|
"Liquidity & Solvency",
|
|
"Efficiency Metrics",
|
|
"Market Data (Future)"
|
|
]
|
|
|
|
for category in expected_categories:
|
|
assert category in categories
|
|
|
|
def test_catalog_category_structure(client: TestClient):
|
|
"""Test catalog category structure"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
categories = data["categories"]
|
|
|
|
# Check each category has proper structure
|
|
for category_name, items in categories.items():
|
|
assert isinstance(items, list)
|
|
assert len(items) > 0
|
|
|
|
# Check each item structure
|
|
for item in items:
|
|
assert "field_name" in item
|
|
assert "description" in item
|
|
assert "data_type" in item
|
|
assert "source" in item
|
|
|
|
# Optional fields
|
|
assert "unit" in item # Can be null
|
|
assert "calculation" in item # Can be null
|
|
|
|
# Check data types
|
|
assert isinstance(item["field_name"], str)
|
|
assert isinstance(item["description"], str)
|
|
assert isinstance(item["data_type"], str)
|
|
assert isinstance(item["source"], str)
|
|
|
|
def test_catalog_specific_fields(client: TestClient):
|
|
"""Test specific fields in catalog"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
categories = data["categories"]
|
|
|
|
# Check some specific fields exist
|
|
company_info = categories["Company Information"]
|
|
field_names = [item["field_name"] for item in company_info]
|
|
|
|
assert "ticker" in field_names
|
|
assert "name" in field_names
|
|
assert "cik" in field_names
|
|
assert "sector" in field_names
|
|
assert "industry" in field_names
|
|
|
|
# Check income statement fields
|
|
income_statement = categories["Income Statement"]
|
|
income_fields = [item["field_name"] for item in income_statement]
|
|
|
|
assert "revenue" in income_fields
|
|
assert "gross_profit" in income_fields
|
|
assert "operating_income" in income_fields
|
|
assert "net_income" in income_fields
|
|
assert "eps" in income_fields
|
|
|
|
def test_catalog_valuation_ratios(client: TestClient):
|
|
"""Test valuation ratios in catalog"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
categories = data["categories"]
|
|
valuation_ratios = categories["Valuation Ratios"]
|
|
|
|
ratio_names = [item["field_name"] for item in valuation_ratios]
|
|
|
|
# Check key ratios exist
|
|
assert "pe_ratio" in ratio_names
|
|
assert "pb_ratio" in ratio_names
|
|
assert "ps_ratio" in ratio_names
|
|
assert "ev_ebitda" in ratio_names
|
|
|
|
# Check they have calculations
|
|
for item in valuation_ratios:
|
|
if item["field_name"] in ["pe_ratio", "pb_ratio", "ps_ratio", "ev_ebitda"]:
|
|
assert item["calculation"] is not None
|
|
assert len(item["calculation"]) > 0
|
|
|
|
def test_catalog_profitability_metrics(client: TestClient):
|
|
"""Test profitability metrics in catalog"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
categories = data["categories"]
|
|
profitability = categories["Profitability Metrics"]
|
|
|
|
profit_names = [item["field_name"] for item in profitability]
|
|
|
|
# Check key profitability metrics
|
|
assert "roe" in profit_names
|
|
assert "roa" in profit_names
|
|
assert "gross_margin" in profit_names
|
|
assert "operating_margin" in profit_names
|
|
assert "net_margin" in profit_names
|
|
|
|
# Check they have percentage units
|
|
for item in profitability:
|
|
if "margin" in item["field_name"] or item["field_name"] in ["roe", "roa"]:
|
|
assert item["unit"] == "percentage"
|
|
|
|
def test_catalog_market_data_future(client: TestClient):
|
|
"""Test future market data fields in catalog"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
categories = data["categories"]
|
|
market_data = categories["Market Data (Future)"]
|
|
|
|
market_names = [item["field_name"] for item in market_data]
|
|
|
|
# Check future fields exist
|
|
assert "market_cap" in market_names
|
|
assert "forward_pe" in market_names
|
|
assert "peg_ratio" in market_names
|
|
assert "beta" in market_names
|
|
|
|
# Check they indicate external data requirement
|
|
for item in market_data:
|
|
if item["field_name"] in ["forward_pe", "peg_ratio", "beta"]:
|
|
assert "external" in item["source"].lower() or "required" in item["source"].lower()
|
|
|
|
def test_catalog_response_format(client: TestClient):
|
|
"""Test catalog response format"""
|
|
response = client.get("/api/v1/metadata/catalog")
|
|
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/json"
|
|
|
|
data = response.json()
|
|
|
|
# Check timestamp format
|
|
assert "last_updated" in data
|
|
assert isinstance(data["last_updated"], str)
|
|
|
|
# Should be ISO format datetime
|
|
from datetime import datetime
|
|
try:
|
|
datetime.fromisoformat(data["last_updated"].replace('Z', '+00:00'))
|
|
except ValueError:
|
|
pytest.fail("last_updated is not in valid ISO format") |