feat: add test framework, unit tests, and GitLab CI pipeline

- Add pytest config in backend/pyproject.toml with asyncio mode and coverage
- Add backend/requirements-test.txt with pytest, pytest-cov, pytest-asyncio, aiosqlite
- Write 86 backend unit tests covering auth (JWT tokens), schemas (Pydantic
  models), config (Settings env overrides), log_parser (log parsing, IP
  resolution, geo data), models (SQLAlchemy table definitions + column
  defaults), and theme export
- Add Vitest config for Angular frontend (vitest.config.ts, src/test-setup.ts)
- Add .gitlab-ci.yml with stages: backend-test, frontend-test, backend-lint,
  docker-build, and quality-gate
- CI enforces 10% minimum coverage threshold for backend
This commit is contained in:
Hermes Agent
2026-07-29 08:34:42 +00:00
parent 8d233b17a5
commit 501df79e5c
12 changed files with 887 additions and 0 deletions
+21
View File
@@ -0,0 +1,21 @@
"""Shared test fixtures for kmtnflower backend tests."""
import pytest
from unittest.mock import patch
# Override settings for tests — use SQLite so we don't need PostgreSQL
@pytest.fixture(autouse=True)
def override_settings():
"""Set test-only environment variables before any import of app modules."""
env = {
"KMTN_DATABASE_URL": "sqlite+aiosqlite:///:memory:",
"KMTN_ENV": "test",
"KMTN_JWT_SECRET_KEY": "test-secret-key-do-not-use",
"KMTN_ADMIN_USERNAME": "testadmin",
"KMTN_ADMIN_PASSWORD": "testpass123",
"KMTN_LOGS_DIR": "/tmp/kmtn_test_logs",
"KMTN_GEOLITE2_DB_PATH": "/nonexistent/GeoLite2-City.mmdb",
"KMTN_THEME_JSON_PATH": "/tmp/kmtn_test_theme.json",
}
with patch.dict("os.environ", env, clear=False):
yield env
+141
View File
@@ -0,0 +1,141 @@
"""Tests for app.auth — JWT token creation, verification, and decoding."""
import pytest
from datetime import datetime, timedelta, timezone
from fastapi import HTTPException
from jose import jwt
from app.auth import (
create_access_token,
create_download_token,
verify_download_token,
decode_token,
)
from app.config import settings
class TestCreateAccessToken:
"""Tests for create_access_token()."""
def test_returns_string_token(self):
token = create_access_token(user_id=1, is_admin=True)
assert isinstance(token, str)
assert len(token) > 0
def test_token_contains_correct_payload(self):
token = create_access_token(user_id=42, is_admin=False)
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
assert payload["sub"] == "42"
assert payload["is_admin"] is False
assert "exp" in payload
def test_admin_flag_is_preserved(self):
admin_token = create_access_token(user_id=1, is_admin=True)
payload = jwt.decode(admin_token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
assert payload["is_admin"] is True
def test_token_expires_in_configured_minutes(self):
token = create_access_token(user_id=1, is_admin=True)
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
expected_exp = datetime.now(timezone.utc) + timedelta(minutes=settings.JWT_EXPIRE_MINUTES)
# Allow 1s drift
assert abs(payload["exp"] - expected_exp.timestamp()) < 2
class TestCreateDownloadToken:
"""Tests for create_download_token()."""
def test_download_token_contains_artifact_id(self):
token = create_download_token(artifact_id=99)
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
assert payload["sub"] == "download"
assert payload["artifact_id"] == 99
def test_download_token_expires_quickly(self):
token = create_download_token(artifact_id=1)
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
expected_exp = datetime.now(timezone.utc) + timedelta(minutes=settings.DOWNLOAD_TOKEN_MINUTES)
assert abs(payload["exp"] - expected_exp.timestamp()) < 2
class TestVerifyDownloadToken:
"""Tests for verify_download_token()."""
def test_valid_download_token_returns_artifact_id(self):
token = create_download_token(artifact_id=55)
result = verify_download_token(token)
assert result == 55
def test_access_token_rejected_as_download_token(self):
token = create_access_token(user_id=1, is_admin=True)
result = verify_download_token(token)
assert result is None
def test_expired_token_returns_none(self):
payload = {
"sub": "download",
"artifact_id": 1,
"exp": datetime.now(timezone.utc) - timedelta(minutes=1),
}
expired_token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
result = verify_download_token(expired_token)
assert result is None
def test_tampered_token_returns_none(self):
token = create_download_token(artifact_id=10)
tampered = token[:-5] + "xxxxx"
result = verify_download_token(tampered)
assert result is None
def test_wrong_secret_returns_none(self):
payload = {
"sub": "download",
"artifact_id": 1,
"exp": datetime.now(timezone.utc) + timedelta(minutes=5),
}
wrong_token = jwt.encode(payload, "wrong-secret", algorithm="HS256")
result = verify_download_token(wrong_token)
assert result is None
def test_non_integer_artifact_id_returns_none(self):
payload = {
"sub": "download",
"artifact_id": "not_an_int",
"exp": datetime.now(timezone.utc) + timedelta(minutes=5),
}
token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
result = verify_download_token(token)
assert result is None
class TestDecodeToken:
"""Tests for decode_token()."""
def test_valid_token_returns_payload(self):
token = create_access_token(user_id=7, is_admin=False)
payload = decode_token(token)
assert payload["sub"] == "7"
def test_expired_token_raises_401(self):
payload = {
"sub": "1",
"exp": datetime.now(timezone.utc) - timedelta(hours=1),
}
expired = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
with pytest.raises(HTTPException) as exc_info:
decode_token(expired)
assert exc_info.value.status_code == 401
def test_invalid_signature_raises_401(self):
with pytest.raises(HTTPException) as exc_info:
decode_token("not.a.token")
assert exc_info.value.status_code == 401
def test_token_without_sub_raises_401(self):
payload = {
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
}
token = jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256")
with pytest.raises(HTTPException) as exc_info:
decode_token(token)
assert exc_info.value.status_code == 401
+117
View File
@@ -0,0 +1,117 @@
"""Tests for app.config — Settings model and environment variable overrides."""
import os
import pytest
from app.config import Settings
def _clear_kmn_env(monkeypatch):
"""Remove all KMTN_ prefixed env vars so Settings reads pure defaults."""
keys = [k for k in os.environ if k.startswith("KMTN_")]
for k in keys:
monkeypatch.delenv(k, raising=False)
class TestSettingsDefaults:
"""Verify Settings defaults match documented behavior."""
def test_default_database_url(self, monkeypatch):
_clear_kmn_env(monkeypatch)
settings = Settings()
assert "postgresql" in settings.DATABASE_URL
assert "kmountain" in settings.DATABASE_URL
def test_default_env_is_development(self, monkeypatch):
_clear_kmn_env(monkeypatch)
settings = Settings()
assert settings.ENV == "development"
def test_default_jwt_expire_minutes(self, monkeypatch):
_clear_kmn_env(monkeypatch)
settings = Settings()
assert settings.JWT_EXPIRE_MINUTES == 1440
def test_default_download_token_minutes(self, monkeypatch):
_clear_kmn_env(monkeypatch)
settings = Settings()
assert settings.DOWNLOAD_TOKEN_MINUTES == 10
def test_default_cors_origins(self, monkeypatch):
_clear_kmn_env(monkeypatch)
settings = Settings()
assert "http://localhost:4200" in settings.CORS_ORIGINS
def test_empty_admin_credentials_by_default(self, monkeypatch):
_clear_kmn_env(monkeypatch)
settings = Settings()
assert settings.ADMIN_USERNAME == ""
assert settings.ADMIN_PASSWORD == ""
class TestSettingsEnvOverride:
"""Verify KMTN_ prefixed env vars override defaults."""
def test_database_url_from_env(self, monkeypatch):
monkeypatch.setenv("KMTN_DATABASE_URL", "sqlite+aiosqlite:///test.db")
settings = Settings()
assert settings.DATABASE_URL == "sqlite+aiosqlite:///test.db"
def test_env_from_env(self, monkeypatch):
monkeypatch.setenv("KMTN_ENV", "production")
settings = Settings()
assert settings.ENV == "production"
def test_jwt_secret_from_env(self, monkeypatch):
monkeypatch.setenv("KMTN_JWT_SECRET_KEY", "my-custom-secret")
settings = Settings()
assert settings.JWT_SECRET_KEY == "my-custom-secret"
def test_jwt_expire_override(self, monkeypatch):
monkeypatch.setenv("KMTN_JWT_EXPIRE_MINUTES", "60")
settings = Settings()
assert settings.JWT_EXPIRE_MINUTES == 60
def test_logs_dir_override(self, monkeypatch):
monkeypatch.setenv("KMTN_LOGS_DIR", "/custom/logs")
settings = Settings()
assert settings.LOGS_DIR == "/custom/logs"
def test_geolite2_path_override(self, monkeypatch):
monkeypatch.setenv("KMTN_GEOLITE2_DB_PATH", "/custom/geo.mmdb")
settings = Settings()
assert settings.GEOLITE2_DB_PATH == "/custom/geo.mmdb"
def test_theme_json_path_override(self, monkeypatch):
monkeypatch.setenv("KMTN_THEME_JSON_PATH", "/custom/theme.json")
settings = Settings()
assert settings.THEME_JSON_PATH == "/custom/theme.json"
def test_admin_credentials_from_env(self, monkeypatch):
monkeypatch.setenv("KMTN_ADMIN_USERNAME", "admin")
monkeypatch.setenv("KMTN_ADMIN_PASSWORD", "secure123")
settings = Settings()
assert settings.ADMIN_USERNAME == "admin"
assert settings.ADMIN_PASSWORD == "secure123"
def test_website_url_override(self, monkeypatch):
monkeypatch.setenv("KMTN_WEBSITE_URL", "https://example.com")
settings = Settings()
assert settings.WEBSITE_URL == "https://example.com"
def test_static_upstream_url_override(self, monkeypatch):
monkeypatch.setenv("KMTN_STATIC_UPSTREAM_URL", "http://frontend:80")
settings = Settings()
assert settings.STATIC_UPSTREAM_URL == "http://frontend:80"
def test_cors_origins_override(self, monkeypatch):
# pydantic-settings parses JSON lists for list[str] fields
monkeypatch.setenv("KMTN_CORS_ORIGINS", '["https://example.com"]')
settings = Settings()
assert settings.CORS_ORIGINS == ["https://example.com"]
def test_kmn_prefix_is_required(self, monkeypatch):
"""Non-prefixed env vars should NOT override settings."""
monkeypatch.setenv("JWT_SECRET_KEY", "should-not-apply")
settings = Settings()
assert settings.JWT_SECRET_KEY != "should-not-apply"
+140
View File
@@ -0,0 +1,140 @@
"""Tests for log_parser — IP resolution, log line parsing, and geo lookups."""
import os
import tempfile
import pytest
from app.log_parser import parse_log_line, resolve_client_ip, _COUNTRY_CENTROIDS
class TestParseLogLine:
"""Tests for parse_log_line()."""
def test_valid_json_line(self):
line = '{"remote_addr": "192.168.1.1", "method": "GET"}'
result = parse_log_line(line)
assert result is not None
assert result["remote_addr"] == "192.168.1.1"
assert result["method"] == "GET"
def test_empty_line_returns_none(self):
assert parse_log_line("") is None
def test_whitespace_only_returns_none(self):
assert parse_log_line(" ") is None
def test_invalid_json_returns_none(self):
assert parse_log_line("not json at all") is None
def test_partial_json_returns_none(self):
assert parse_log_line('{"remote_addr": ') is None
def test_strips_whitespace(self):
line = ' {"key": "value"} \n'
result = parse_log_line(line)
assert result is not None
assert result["key"] == "value"
class TestResolveClientIp:
"""Tests for resolve_client_ip()."""
def test_remote_addr_without_xff(self):
entry = {"remote_addr": "203.0.113.5"}
assert resolve_client_ip(entry) == "203.0.113.5"
def test_xff_single_ip(self):
entry = {
"remote_addr": "10.0.0.1",
"x_forwarded_for": "203.0.113.50",
}
assert resolve_client_ip(entry) == "203.0.113.50"
def test_xff_multiple_ips_returns_leftmost(self):
entry = {
"remote_addr": "10.0.0.1",
"x_forwarded_for": "203.0.113.100, 10.1.2.3, 172.16.0.5",
}
assert resolve_client_ip(entry) == "203.0.113.100"
def test_xff_dash_falls_back_to_remote_addr(self):
entry = {"remote_addr": "192.168.0.1", "x_forwarded_for": "-"}
assert resolve_client_ip(entry) == "192.168.0.1"
def test_missing_fields_returns_empty(self):
entry = {}
assert resolve_client_ip(entry) == ""
def test_xff_with_whitespace(self):
entry = {
"remote_addr": "10.0.0.1",
"x_forwarded_for": " 203.0.113.75 ",
}
assert resolve_client_ip(entry) == "203.0.113.75"
class TestCountryCentroids:
"""Tests for _COUNTRY_CENTROIDS data integrity."""
def test_centroids_are_float_tuples(self):
for code, coords in _COUNTRY_CENTROIDS.items():
assert isinstance(coords, tuple)
assert len(coords) == 2
assert isinstance(coords[0], float)
assert isinstance(coords[1], float)
def test_known_countries_present(self):
expected = {"US", "GB", "DE", "FR", "JP", "AU", "BR", "CA", "IN"}
assert expected.issubset(_COUNTRY_CENTROIDS.keys())
def test_centroid_values_are_reasonable(self):
"""Latitudes should be in [-90, 90], longitudes in [-180, 180]."""
for code, (lat, lon) in _COUNTRY_CENTROIDS.items():
assert -90 <= lat <= 90, f"{code} latitude {lat} out of range"
assert -180 <= lon <= 180, f"{code} longitude {lon} out of range"
class TestParseLogFile:
"""Tests for parse_log_file() — reads from real temp files."""
def test_parses_multiple_lines(self, tmp_path):
log_content = (
'{"remote_addr": "1.2.3.4"}\n'
'{"remote_addr": "5.6.7.8"}\n'
'{"remote_addr": "1.2.3.4"}\n'
)
log_file = tmp_path / "test.log"
log_file.write_text(log_content)
from app.log_parser import parse_log_file
result = parse_log_file(str(log_file))
# Should have 2 unique IPs (1.2.3.4 appears twice)
assert "1.2.3.4" in result["unique_ips"]
assert "5.6.7.8" in result["unique_ips"]
assert len(result["unique_ips"]) == 2
def test_skips_invalid_lines(self, tmp_path):
log_content = (
'{"remote_addr": "1.2.3.4"}\n'
'not valid json\n'
'\n'
'{"remote_addr": "5.6.7.8"}\n'
)
log_file = tmp_path / "test.log"
log_file.write_text(log_content)
from app.log_parser import parse_log_file
result = parse_log_file(str(log_file))
assert len(result["unique_ips"]) == 2
def test_xff_takes_precedence(self, tmp_path):
log_content = (
'{"remote_addr": "10.0.0.1", "x_forwarded_for": "203.0.113.1"}\n'
)
log_file = tmp_path / "test.log"
log_file.write_text(log_content)
from app.log_parser import parse_log_file
result = parse_log_file(str(log_file))
assert "203.0.113.1" in result["unique_ips"]
assert "10.0.0.1" not in result["unique_ips"]
+111
View File
@@ -0,0 +1,111 @@
"""Tests for app.models — SQLAlchemy model definitions and relationships."""
import pytest
from app.models import (
Base,
Show,
ShowSchedule,
Event,
StationConfig,
HistoryEntry,
TeamMember,
CommunityHighlight,
Underwriter,
VisitorStatDaily,
VisitorStatGeo,
)
class TestModelDefinitions:
"""Verify that models are properly defined."""
def test_show_table_name(self):
assert Show.__tablename__ == "shows"
def test_show_schedule_table_name(self):
assert ShowSchedule.__tablename__ == "show_schedules"
def test_event_table_name(self):
assert Event.__tablename__ == "events"
def test_station_config_table_name(self):
assert StationConfig.__tablename__ == "station_config"
def test_history_entry_table_name(self):
assert HistoryEntry.__tablename__ == "history_entries"
def test_team_member_table_name(self):
assert TeamMember.__tablename__ == "team_members"
def test_community_highlight_table_name(self):
assert CommunityHighlight.__tablename__ == "community_highlights"
def test_underwriter_table_name(self):
assert Underwriter.__tablename__ == "underwriters"
def test_visitor_stats_daily_table_name(self):
assert VisitorStatDaily.__tablename__ == "visitor_stats_daily"
def test_visitor_stats_geo_table_name(self):
assert VisitorStatGeo.__tablename__ == "visitor_stats_geo"
class TestModelDefaults:
"""Verify SQLAlchemy column-level defaults (set at INSERT time)."""
def _col_default_value(self, model_cls, col_name):
"""Extract the Column default value from a model's metadata."""
col = model_cls.__table__.columns[col_name]
default = col.default
if default is None:
return None
# DefaultClause.arg is a property (not callable) in SQLAlchemy 2.0
if hasattr(default, "arg"):
return default.arg
return default
def test_show_defaults(self):
assert self._col_default_value(Show, "display_order") == 0
assert self._col_default_value(Show, "active") is True
def test_event_defaults(self):
assert self._col_default_value(Event, "active") is True
def test_station_config_defaults(self):
assert self._col_default_value(StationConfig, "callsign") == "KMTN"
assert self._col_default_value(StationConfig, "name_primary") == "KMountain"
assert self._col_default_value(StationConfig, "name_secondary") == "Flower Radio"
assert self._col_default_value(StationConfig, "frequency") == "98.7 FM"
assert self._col_default_value(StationConfig, "founded_year") == 1987
assert self._col_default_value(StationConfig, "nonprofit_type") == "501(c)(3)"
def test_history_entry_defaults(self):
assert self._col_default_value(HistoryEntry, "active") is True
def test_team_member_defaults(self):
assert self._col_default_value(TeamMember, "active") is True
def test_underwriter_defaults(self):
assert self._col_default_value(Underwriter, "display_order") == 0
assert self._col_default_value(Underwriter, "active") is True
class TestBaseModel:
"""Verify Base declarative model setup."""
def test_base_has_metadata(self):
assert hasattr(Base, "metadata")
def test_all_models_registered_in_metadata(self):
tables = Base.metadata.tables
assert "shows" in tables
assert "show_schedules" in tables
assert "events" in tables
assert "station_config" in tables
assert "history_entries" in tables
assert "team_members" in tables
assert "community_highlights" in tables
assert "underwriters" in tables
assert "visitor_stats_daily" in tables
assert "visitor_stats_geo" in tables
+177
View File
@@ -0,0 +1,177 @@
"""Tests for Pydantic schemas — validation, defaults, and computed fields."""
from datetime import date
import pytest
from app.schemas import (
EventCreate,
EventUpdate,
EventResponse,
StationConfigUpdate,
ShowScheduleResponse,
ShowCreate,
ShowScheduleCreate,
HistoryEntryCreate,
TeamMemberCreate,
UnderwriterCreate,
DAY_NAMES,
)
class TestEventSchemas:
"""Tests for Event Pydantic schemas."""
def test_event_create_validates_required_fields(self):
event = EventCreate(
date=date(2026, 8, 15),
title="Summer Concert",
description="Live music",
location="Main Stage",
icon="🎵",
)
assert event.title == "Summer Concert"
assert event.rsvp_url is None
def test_event_create_with_rsvp(self):
event = EventCreate(
date=date(2026, 9, 1),
title="Festival",
description="Big event",
location="Park",
icon="🎉",
rsvp_url="https://example.com/rsvp",
)
assert event.rsvp_url == "https://example.com/rsvp"
def test_event_update_all_optional(self):
update = EventUpdate()
assert update.title is None
assert update.active is None
def test_event_update_partial(self):
update = EventUpdate(active=False, title="Updated Title")
assert update.active is False
assert update.title == "Updated Title"
assert update.date is None
class TestStationConfigUpdate:
"""Tests for StationConfigUpdate schema."""
def test_all_fields_optional(self):
update = StationConfigUpdate()
assert update.callsign is None
assert update.color_primary is None
def test_partial_update(self):
update = StationConfigUpdate(
callsign="KXYZ",
color_primary="#ff0000",
)
assert update.callsign == "KXYZ"
assert update.color_primary == "#ff0000"
assert update.name_primary is None
class TestShowScheduleResponse:
"""Tests for ShowScheduleResponse computed field."""
def test_day_label_mapping(self):
for day_num, expected_label in DAY_NAMES.items():
schedule = ShowScheduleResponse(
id=1,
show_id=1,
day_of_week=day_num,
time="10:00",
)
assert schedule.day_label == expected_label
def test_unknown_day_returns_unknown(self):
schedule = ShowScheduleResponse(
id=1,
show_id=1,
day_of_week=99,
time="10:00",
)
assert schedule.day_label == "Unknown"
class TestShowCreate:
"""Tests for ShowCreate schema."""
def test_show_create_with_schedules(self):
show = ShowCreate(
title="Morning Show",
description="Morning music",
host="DJ Alex",
genre="Pop",
schedules=[
ShowScheduleCreate(day_of_week=1, time="08:00"),
ShowScheduleCreate(day_of_week=3, time="08:00"),
],
)
assert len(show.schedules) == 2
assert show.schedules[0].day_of_week == 1
assert show.display_order == 0
def test_show_create_defaults(self):
show = ShowCreate(
title="Show",
description="Desc",
host="Host",
genre="Genre",
schedules=[ShowScheduleCreate(day_of_week=1, time="09:00")],
)
assert show.show_art_url is None
assert show.hero_image_url is None
assert show.display_order == 0
class TestHistoryEntryCreate:
"""Tests for HistoryEntryCreate schema."""
def test_valid_creation(self):
entry = HistoryEntryCreate(
year=1987,
title="Station Founded",
body="The station began broadcasting.",
display_order=1,
)
assert entry.year == 1987
assert entry.title == "Station Founded"
class TestTeamMemberCreate:
"""Tests for TeamMemberCreate schema."""
def test_required_fields(self):
member = TeamMemberCreate(
name="Jane Doe",
title="Engineer",
display_order=1,
)
assert member.name == "Jane Doe"
assert member.bio is None
assert member.photo_url is None
class TestUnderwriterCreate:
"""Tests for UnderwriterCreate schema."""
def test_defaults_display_order_to_zero(self):
uw = UnderwriterCreate(
name="Sponsor Co",
description="A great sponsor",
)
assert uw.display_order == 0
assert uw.website_url is None
assert uw.logo_url is None
def test_custom_display_order(self):
uw = UnderwriterCreate(
name="Premium Sponsor",
description="Best sponsor",
display_order=5,
website_url="https://sponsor.example.com",
)
assert uw.display_order == 5
+22
View File
@@ -0,0 +1,22 @@
"""Tests for theme_export — theme.json generation from station config."""
import os
import json
import pytest
from unittest.mock import patch, AsyncMock, MagicMock
from app.config import Settings
class TestThemeExport:
"""Tests for theme.json export logic."""
def test_settings_theme_json_path_default(self):
with patch.dict("os.environ", {"KMTN_THEME_JSON_PATH": "/app/theme.json"}, clear=False):
settings = Settings()
assert settings.THEME_JSON_PATH == "/app/theme.json"
def test_settings_website_url_default(self):
with patch.dict("os.environ", {"KMTN_WEBSITE_URL": "https://kmountainflower.org"}, clear=False):
settings = Settings()
assert settings.WEBSITE_URL == "https://kmountainflower.org"