from datetime import date from pathlib import Path import pandas as pd from flask import Flask from sqlalchemy import create_engine, func from sqlalchemy.orm import Session from database.models import MonthlySummaryAnalysis from routes.daily_sales_routes import prepare_daily_charts import routes.daily_sales_routes as daily_routes from routes.monthly_routes import ( _apply_monthly_period, _resolve_monthly_period, get_monthly_summary_data, ) import routes.sales_routes as sales_routes from routes.sales_routes import ( _filter_growth_payload, _period_month_keys, _project_yoy_period, _sorted_valid_month_labels, ) from services.analysis_period_service import make_analysis_period, make_month_period ROOT = Path(__file__).resolve().parents[1] def test_daily_month_charts_and_top_products_use_the_selected_month_only(): frame = pd.DataFrame([ {"snapshot_date": "2026-04-01", "商品名稱": "A", "總業績": 100, "總成本": 60, "數量": 1}, {"snapshot_date": "2026-04-30", "商品名稱": "B", "總業績": 300, "總成本": 180, "數量": 3}, {"snapshot_date": "2026-07-01", "商品名稱": "C", "總業績": 9999, "總成本": 1, "數量": 1}, ]) frame["snapshot_date"] = pd.to_datetime(frame["snapshot_date"]) charts = prepare_daily_charts( frame, pd.Timestamp("2026-04-30"), start_date=pd.Timestamp("2026-04-01"), end_date=pd.Timestamp("2026-04-30"), top10_scope="range", ) assert charts["labels"] == ["04/01", "04/30"] assert charts["revenue"] == [100, 300] assert charts["top10_labels"] == ["B", "A"] assert charts["top10_values"] == [300, 100] def test_daily_page_month_query_drives_kpis_charts_and_tables_from_same_month(monkeypatch): engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.exec_driver_sql( """ CREATE TABLE daily_sales_snapshot ( snapshot_date TEXT, "商品名稱" TEXT, "館別" TEXT, "總業績" TEXT, "總成本" TEXT, "數量" TEXT ) """ ) conn.exec_driver_sql( """ INSERT INTO daily_sales_snapshot (snapshot_date, "商品名稱", "館別", "總業績", "總成本", "數量") VALUES ('2026-04-01', 'A', '美妝', '100', '60', '1'), ('2026-04-30', 'B', '保養', '300', '180', '3'), ('2026-07-01', 'C', '3C', '9999', '1', '1') """ ) class FakeDatabaseManager: def __init__(self): self.engine = engine monkeypatch.setattr(daily_routes, "DatabaseManager", FakeDatabaseManager) monkeypatch.setattr(daily_routes, "_get_shared_daily_view_cache", lambda _key: None) monkeypatch.setattr(daily_routes, "_set_shared_daily_view_cache", lambda _key, _context: None) monkeypatch.setattr( daily_routes, "build_competitor_intel_payload", lambda *_args, **_kwargs: {"trend": {"labels": []}, "snapshot_in_period": False}, ) monkeypatch.setattr(daily_routes, "render_template", lambda _name, **context: context) daily_routes._DAILY_SALES_VIEW_CACHE.clear() daily_routes._SALES_PROCESSED_CACHE.clear() app = Flask(__name__) app.register_blueprint(daily_routes.daily_sales_bp) with app.test_request_context("/daily_sales?month=2026-04"): context = daily_routes.daily_sales.__wrapped__() assert context["selected_date"] == "2026-04-30" assert context["analysis_period"]["start_date"] == "2026-04-01" assert context["analysis_period"]["end_date"] == "2026-04-30" assert context["month_kpi"]["total_revenue"] == 400 assert context["chart_data"]["labels"] == ["04/01", "04/30"] assert {item["category"] for item in context["categories"]} == {"美妝", "保養"} def test_canonical_period_contract_exposes_day_month_and_range_keys(): month = make_month_period("2026-04") custom = make_analysis_period(date(2026, 3, 15), date(2026, 4, 20)) assert month == { "mode": "month", "label": "2026 年 04 月", "start_date": "2026-04-01", "end_date": "2026-04-30", "start_month": "2026-04", "end_month": "2026-04", "is_single_day": False, "is_single_month": True, } assert custom["start_month"] == "2026-03" assert custom["end_month"] == "2026-04" assert custom["label"] == "2026-03-15 至 2026-04-20" def test_monthly_period_query_includes_only_current_and_aligned_previous_period(): engine = create_engine("sqlite:///:memory:") MonthlySummaryAnalysis.__table__.create(engine) with Session(engine) as session: session.add_all([ MonthlySummaryAnalysis(year=2024, month=4, sales_amt_curr=100), MonthlySummaryAnalysis(year=2025, month=4, sales_amt_curr=200), MonthlySummaryAnalysis(year=2026, month=4, sales_amt_curr=300), ]) session.commit() period = _resolve_monthly_period(session, start_month="2026-04", end_month="2026-04") month_only_period = _resolve_monthly_period(session, month=4) current = _apply_monthly_period( session.query(func.sum(MonthlySummaryAnalysis.sales_amt_curr)), period, ).scalar() compared = _apply_monthly_period( session.query(MonthlySummaryAnalysis.year).order_by(MonthlySummaryAnalysis.year), period, include_previous=True, ).all() assert period["label"] == "2026 年 04 月" assert period["previous_label"] == "2025 年 04 月" assert month_only_period["label"] == "2026 年 04 月" assert current == 300 assert [row.year for row in compared] == [2025, 2026] def test_monthly_api_links_kpis_tables_and_comparison_charts_to_one_period(monkeypatch): engine = create_engine("sqlite:///:memory:") MonthlySummaryAnalysis.__table__.create(engine) with Session(engine) as session: session.add_all([ MonthlySummaryAnalysis( year=2025, month=4, division="美妝", area_name="臉部保養", brand_name="A", vendor_name="Vendor A", trade_type="B2C", unit_price=100, sales_amt_curr=200, sales_amt_prev=180, sales_amt_yoa=150, profit_amt_curr=40, sales_vol_curr=2, views_curr=20, ), MonthlySummaryAnalysis( year=2026, month=4, division="美妝", area_name="臉部保養", brand_name="A", vendor_name="Vendor A", trade_type="B2C", unit_price=100, sales_amt_curr=300, sales_amt_prev=250, sales_amt_yoa=200, profit_amt_curr=60, sales_vol_curr=3, views_curr=30, ), ]) session.commit() class FakeDatabaseManager: def __init__(self): self.engine = engine def get_session(self): return Session(engine) monkeypatch.setattr("routes.monthly_routes.DatabaseManager", FakeDatabaseManager) monkeypatch.setattr("routes.monthly_routes.DATABASE_TYPE", "sqlite") app = Flask(__name__) with app.test_request_context( "/api/monthly_summary_data?start_month=2026-04&end_month=2026-04" ): response = get_monthly_summary_data.__wrapped__() payload = response.get_json() assert payload["status"] == "success" assert payload["period"]["label"] == "2026 年 04 月" assert payload["kpis"]["sales"] == 300 assert payload["total_rows"] == 1 assert [(item["date"], item["sales"]) for item in payload["trend"]] == [ ("2025/4", 200), ("2026/4", 300), ] assert payload["vendor_ranking"][0]["sales_previous"] == 200 assert payload["vendor_ranking"][0]["sales_current"] == 300 assert {row["year"] for row in payload["rows"]} == {2026} def test_growth_period_slices_every_series_and_recomputes_period_kpis(): chart_data = { "labels": ["2025-04", "2026-03", "2026-04", "2026-05"], "revenue": [100, 200, 300, 400], "profit": [10, 20, 30, 40], "orders": [1, 2, 3, 4], "aov": [100, 100, 100, 100], "mom": [0, 100, 50, 33.3], "yoy": [0, 0, 200, 0], "margin_rate": [10, 10, 10, 10], "competitor_gap_pct": [1, 2, 3, 4], "competitor_coverage": {"decision_support_rate": 25}, } filtered, kpi, period, is_filtered = _filter_growth_payload( chart_data, {"ytd_revenue": 900, "total_orders": 10}, "2026-04", "2026-05", ) assert is_filtered is True assert filtered["labels"] == ["2026-04", "2026-05"] assert filtered["revenue"] == [300, 400] assert filtered["profit"] == [30, 40] assert filtered["competitor_gap_pct"] == [3, 4] assert filtered["competitor_snapshot_in_period"] is False assert filtered["competitor_coverage"] == {} assert kpi["ytd_revenue"] == 700 assert kpi["previous_period_revenue"] == 100 assert kpi["total_orders"] == 7 assert period["start_month"] == "2026-04" assert period["end_month"] == "2026-05" def test_sales_seasonality_month_axis_ignores_nan_and_invalid_values(): assert _sorted_valid_month_labels([ "2026-04", float("nan"), None, "invalid", "2025-12", "2026-04-30", ]) == ["2025-12", "2026-04"] def test_monthly_frontend_has_replayable_period_and_no_fixed_years(): script = (ROOT / "web/static/js/page-monthly-summary.js").read_text(encoding="utf-8") route = (ROOT / "routes/monthly_routes.py").read_text(encoding="utf-8") tabs = (ROOT / "templates/components/_analysis_report_tabs.html").read_text(encoding="utf-8") assert "buildFilterParams" in script assert "syncFilterUrl" in script assert "start_month" in script assert "end_month" in script assert "areaOverride" in script assert "sales_current" in script assert "sales_previous" in script assert "2024" not in script assert "2025" not in script assert "2024" not in route assert "2025" not in route assert "data-analysis-target" in tabs assert "renderChartEmpty" in script assert "未以其他月份回填" in script def test_sales_frontend_uses_live_period_linked_api_routes(): script = (ROOT / "web/static/js/page-sales-analysis.js").read_text(encoding="utf-8") template = (ROOT / "templates/sales_analysis.html").read_text(encoding="utf-8") assert "buildPeriodApiUrl" in script assert "new URLSearchParams(window.location.search)" in script assert "/api/sales_analysis/yoy_comparison" in script assert "/api/sales_analysis/table_data" in script assert "d.growth_rate" in script assert "d.monthly_breakdown" in script assert "columns," in script assert 'data-field="product_id"' in template assert 'data-field="amount"' in template assert "fetch(`/api/yoy?" not in script assert "'/api/sales_table?'" not in script def test_yoy_period_projects_selected_dates_to_each_comparison_year(): start, end = _project_yoy_period(2025, "2026-04-01", "2026-04-30") assert start == date(2025, 4, 1) assert end == date(2025, 4, 30) assert _period_month_keys(start, end) == ["2025-04"] def test_yoy_api_uses_only_the_selected_period_for_totals_and_chart(monkeypatch): engine = create_engine("sqlite:///:memory:") with engine.begin() as conn: conn.exec_driver_sql( 'CREATE TABLE realtime_sales_monthly ("日期" TEXT, "總業績" REAL, "數量" REAL, "總成本" REAL)' ) conn.exec_driver_sql( """ INSERT INTO realtime_sales_monthly ("日期", "總業績", "數量", "總成本") VALUES ('2025/04/10', 100, 2, 60), ('2025/05/10', 900, 9, 100), ('2026/04/10', 150, 3, 80), ('2026/05/10', 1200, 12, 200) """ ) class FakeDatabaseManager: def __init__(self): self.engine = engine monkeypatch.setattr(sales_routes, "DatabaseManager", FakeDatabaseManager) app = Flask(__name__) with app.test_request_context( "/api/sales_analysis/yoy_comparison" "?year1=2025&year2=2026&metric=revenue" "&start_date=2026-04-01&end_date=2026-04-30" ): response = sales_routes.api_yoy_comparison.__wrapped__() payload = response.get_json() assert payload["period"]["linked"] is True assert payload["period"]["year1"] == { "start_date": "2025-04-01", "end_date": "2025-04-30", } assert payload["year1"]["total"] == 100 assert payload["year2"]["total"] == 150 assert payload["growth_rate"] == 50 assert payload["monthly_breakdown"] == [{ "month": 4, "month_label": "4月", "year1_value": 100, "year2_value": 150, "growth_rate": 50, }]