office_translator/tests/test_translate_routes_helpers.py

"""
Unit tests for the helper functions in routes/translate_routes.py.

These cover the bug fixes that are pure logic (no full HTTP machinery needed):
- Bug 2: _compute_cost_factor / _provider_model (reads ``_model`` AND ``model``)
- Bug 3: _release_quota_if_needed (releases reserved quota on soft-failure)
- Bug 4: _compute_duration_seconds (UTC-aware, never crashes)
- Bug 5: _cleanup_old_jobs snapshots the jobs dict (no "changed size during iteration")
"""

import asyncio
import time
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock

import pytest

from routes import translate_routes as tr


# ---------------------------------------------------------------------------
# Bug 2 — _provider_model / _compute_cost_factor
# ---------------------------------------------------------------------------
class _NewStyleProvider:
    """Mimics services/providers/* classes that store ``self._model``."""

    def __init__(self, model):
        self._model = model


class _LegacyProvider:
    """Mimics legacy services/translation_service classes with ``self.model``."""

    def __init__(self, model):
        self.model = model


class TestProviderModel:
    def test_reads_new_style_private_attr(self):
        assert tr._provider_model(_NewStyleProvider("gpt-4o")) == "gpt-4o"

    def test_reads_legacy_public_attr(self):
        assert tr._provider_model(_LegacyProvider("claude-sonnet-4")) == "claude-sonnet-4"

    def test_prefers_private_when_both_present(self):
        class Both:
            _model = "private-model"
            model = "public-model"

        assert tr._provider_model(Both()) == "private-model"

    def test_none_provider(self):
        assert tr._provider_model(None) == ""

    def test_empty_model(self):
        assert tr._provider_model(_NewStyleProvider("")) == ""


class TestComputeCostFactor:
    def test_claude_is_premium(self):
        assert tr._compute_cost_factor(_NewStyleProvider("anthropic/claude-sonnet-4.6")) == 5

    def test_gpt4_is_premium(self):
        assert tr._compute_cost_factor(_NewStyleProvider("gpt-4o")) == 5

    def test_gpt4o_mini_is_standard(self):
        # Cheap GPT-4 variants must not be billed at the premium factor.
        assert tr._compute_cost_factor(_NewStyleProvider("gpt-4o-mini")) == 1
        assert tr._compute_cost_factor(_NewStyleProvider("gpt-4o-nano")) == 1

    def test_legacy_gpt4o_mini_is_standard(self):
        assert tr._compute_cost_factor(_LegacyProvider("gpt-4o-mini")) == 1

    def test_haiku_is_standard(self):
        assert tr._compute_cost_factor(_NewStyleProvider("anthropic/claude-3-haiku")) == 1

    def test_openrouter_premium_alias_is_premium_without_model(self):
        # Regression for the original bug: model read failed (""), so the
        # premium tier was never detected. The alias must still bump it to 5.
        assert tr._compute_cost_factor(None, "openrouter_premium") == 5

    def test_standard_model(self):
        assert tr._compute_cost_factor(_NewStyleProvider("deepseek-chat")) == 1

    def test_legacy_provider_model_is_read(self):
        # Legacy classes expose .model — must also be billed correctly.
        assert tr._compute_cost_factor(_LegacyProvider("gpt-4o")) == 5


# ---------------------------------------------------------------------------
# Bug 3 — _release_quota_if_needed
# ---------------------------------------------------------------------------
class TestReleaseQuotaIfNeeded:
    @pytest.mark.asyncio
    async def test_releases_when_user_id_and_not_recorded(self):
        # release_translation_quota is invoked via asyncio.to_thread (a worker
        # thread); patching it and asserting the call validates the release path.
        with patch.object(tr, "release_translation_quota") as mock_release:
            await tr._release_quota_if_needed("user-123", usage_recorded=False, job_id="j1")
            mock_release.assert_called_once_with("user-123")

    @pytest.mark.asyncio
    async def test_skips_when_usage_already_recorded(self):
        with patch.object(tr, "release_translation_quota") as mock_release:
            await tr._release_quota_if_needed("user-123", usage_recorded=True, job_id="j1")
            mock_release.assert_not_called()

    @pytest.mark.asyncio
    async def test_skips_when_no_user_id(self):
        with patch.object(tr, "release_translation_quota") as mock_release:
            await tr._release_quota_if_needed(None, usage_recorded=False, job_id="j1")
            mock_release.assert_not_called()

    @pytest.mark.asyncio
    async def test_swallows_release_errors(self):
        with patch.object(tr, "release_translation_quota", side_effect=RuntimeError("db down")):
            # Must not raise even if the release itself fails.
            await tr._release_quota_if_needed("user-123", usage_recorded=False, job_id="j1")


# ---------------------------------------------------------------------------
# Bug 4 — _compute_duration_seconds (UTC-aware, crash-safe)
# ---------------------------------------------------------------------------
class TestComputeDurationSeconds:
    def test_recent_timestamp_returns_positive(self):
        ts = (datetime.now(timezone.utc) - timedelta(seconds=10)).isoformat()
        dur = tr._compute_duration_seconds(ts)
        assert dur >= 9  # allow tiny scheduling slack

    def test_z_suffix_handled(self):
        ts = (datetime.now(timezone.utc) - timedelta(seconds=5)).strftime("%Y-%m-%dT%H:%M:%SZ")
        dur = tr._compute_duration_seconds(ts)
        assert dur >= 0

    def test_malformed_returns_zero(self):
        # Regression for the original bug: a bad timestamp used to crash the
        # success path and flip the job to failed. It must now return 0.
        assert tr._compute_duration_seconds("not-a-date") == 0.0

    def test_empty_returns_zero(self):
        assert tr._compute_duration_seconds("") == 0.0


# ---------------------------------------------------------------------------
# Bug 5 — _cleanup_old_jobs snapshots the dict (no resize-during-iteration)
# ---------------------------------------------------------------------------
class TestCleanupOldJobsSnapshots:
    def test_cleanup_does_not_raise_when_dict_mutated_concurrently(self, monkeypatch):
        # Force cleanup to run now (bypass throttle).
        monkeypatch.setattr(tr, "_last_cleanup_ts", 0.0)
        monkeypatch.setattr(tr, "_CLEANUP_INTERVAL_SECONDS", 0)

        # Two expired jobs.
        old_ts = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat()
        tr._translation_jobs.clear()
        tr._translation_jobs["j1"] = {"status": "completed", "completed_at": old_ts}
        tr._translation_jobs["j2"] = {"status": "failed", "failed_at": old_ts}
        tr._translation_jobs["j3"] = {"status": "running"}  # not expired

        # If cleanup did NOT snapshot, mutating during iteration would raise.
        tr._cleanup_old_jobs()

        assert "j1" not in tr._translation_jobs
        assert "j2" not in tr._translation_jobs
        assert "j3" in tr._translation_jobs

    def teardown_method(self):
        tr._translation_jobs.clear()