Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions locale/themerr.po
Original file line number Diff line number Diff line change
Expand Up @@ -651,27 +651,27 @@ msgstr ""
msgid "No cookies saved"
msgstr ""

#: web/js/config.js:81 web/js/config.js:123
#: web/js/config.js:82 web/js/config.js:124
msgid "Unsaved changes"
msgstr ""

#: web/js/config.js:89
#: web/js/config.js:90
msgid "Discard unsaved settings changes?"
msgstr ""

#: web/js/config.js:123 web/templates/config.html:40
#: web/js/config.js:124 web/templates/config.html:40
msgid "All changes saved."
msgstr ""

#: web/js/config.js:125
#: web/js/config.js:126
msgid "Settings saved. Network changes take effect after restarting Themerr."
msgstr ""

#: web/js/config.js:129
#: web/js/config.js:130
msgid "Unable to save changes."
msgstr ""

#: web/js/config.js:137
#: web/js/config.js:138
msgid "Save your settings changes before changing your password."
msgstr ""

Expand Down Expand Up @@ -1056,8 +1056,8 @@ msgstr ""

#: web/templates/home.html:8
msgid ""
"GitHub check failed. Showing the last known deployment; checks run at "
"most hourly."
"Deployment check failed. Showing the last known deployment; checks run at"
" most hourly."
msgstr ""

#: web/templates/home.html:9
Expand Down
4 changes: 2 additions & 2 deletions src/common/server_ui.py
Original file line number Diff line number Diff line change
Expand Up @@ -261,8 +261,8 @@ def database_status():
Response
Deployment metadata and check timestamps.
"""
from themerr import github_status
return JSONResponse(github_status.publication_status())
from themerr import deployment_status
return JSONResponse(deployment_status.publication_status())


@router.api_route('/activity', methods=['GET', 'HEAD'], name='server_ui.activity', response_model=None)
Expand Down
50 changes: 22 additions & 28 deletions src/themerr/github_status.py → src/themerr/deployment_status.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
"""Persist the latest successful ThemerrDB deployment and an hourly GitHub check limit."""
"""Persist ThemerrDB's published deployment timestamp and an hourly check limit."""

# standard imports
from datetime import datetime, timezone
Expand All @@ -17,11 +17,9 @@
log = logger.get_logger(__name__)
_lock = Lock()
_CACHE_KEY = 'themerrdb_deployment'
_CACHE_VERSION = 2
_CACHE_VERSION = 3
_INTERVAL = 3600
_WORKFLOW_URL = 'https://github.com/LizardByte/ThemerrDB/actions/workflows/pages/pages-build-deployment'
# GitHub's built-in Pages workflow has no YAML filename; use its stable repository workflow ID.
_WORKFLOW_ID = 34813093
_DEPLOYMENT_URL = 'https://app.lizardbyte.dev/ThemerrDB/deployment.json'


def _load() -> dict:
Expand All @@ -47,33 +45,29 @@ def _save(data: dict) -> None:


def _fetch_publication() -> dict:
"""Validate the latest successful Pages run before returning its metadata."""
"""Validate the timestamp published with the deployed database."""
with requests.get(
f'https://api.github.com/repos/LizardByte/ThemerrDB/actions/workflows/{_WORKFLOW_ID}/runs',
params={'status': 'success', 'per_page': 1},
headers={'Accept': 'application/vnd.github+json', 'Cache-Control': 'no-cache'},
_DEPLOYMENT_URL,
headers={
'Accept': 'application/json',
'Cache-Control': 'no-cache',
},
timeout=10, allow_redirects=False,
) as response:
response.raise_for_status()
if response.status_code != 200:
raise ValueError('Unexpected GitHub response')
runs = response.json()['workflow_runs']
if not isinstance(runs, list):
raise ValueError('Invalid GitHub workflow list')
if not runs:
return {}
run = runs[0]
run_id = run['id']
date = datetime.fromisoformat(run['updated_at'])
if (type(run_id) is not int or run_id <= 0 or date.tzinfo is None or
run['workflow_id'] != _WORKFLOW_ID or run['status'] != 'completed' or
run['conclusion'] != 'success'):
raise ValueError('Invalid GitHub deployment metadata')
return {'updated_at': date.astimezone(timezone.utc).isoformat(), 'run_id': run_id}
raise ValueError('Unexpected deployment response')
metadata = response.json()
if not isinstance(metadata, dict):
raise ValueError('Invalid deployment metadata')
date = datetime.fromisoformat(metadata['deployed_at'])
if date.tzinfo is None:
raise ValueError('Deployment timestamp must include a timezone')
return {'updated_at': date.astimezone(timezone.utc).isoformat()}


def publication_status() -> dict:
"""Return the last successful Pages deployment, checking at most once per hour.
"""Return the published database deployment, checking at most once per hour.

Checks are serialized across application threads. The attempt deadline is saved
before the request, so failures and application restarts also respect the limit.
Expand All @@ -82,7 +76,7 @@ def publication_status() -> dict:
Returns
-------
dict
Deployment completion date and workflow link, last attempt and next check timestamps,
Published deployment date and repository link, last attempt and next check timestamps,
and whether the last check failed. Dates use UTC; timestamps use Unix seconds.
"""
with _lock:
Expand All @@ -101,8 +95,8 @@ def publication_status() -> dict:
_save(state)
return {
'updated_at': state.get('updated_at'),
'url': ('https://github.com/LizardByte/ThemerrDB/actions/runs/' + str(state['run_id'])
if state.get('run_id') else _WORKFLOW_URL),
'checked_at': state.get('checked_at'), 'next_check': state.get('next_check'),
'url': 'https://github.com/LizardByte/ThemerrDB',
'checked_at': state.get('checked_at'),
'next_check': state.get('next_check'),
'stale': state.get('stale', True),
}
4 changes: 2 additions & 2 deletions tests/functional/test_asgi.py
Original file line number Diff line number Diff line change
Expand Up @@ -344,8 +344,8 @@ def test_documentation_supplies_session_bound_csrf_defaults(browser):


def test_unexpected_api_failure_keeps_generic_errors_and_security_headers(browser, monkeypatch):
from themerr import github_status
monkeypatch.setattr(github_status, 'publication_status', Mock(side_effect=RuntimeError('private details')))
from themerr import deployment_status
monkeypatch.setattr(deployment_status, 'publication_status', Mock(side_effect=RuntimeError('private details')))
with TestClient(browser.app, base_url='http://localhost', raise_server_exceptions=False,
cookies=browser.cookies) as client:
response = client.get('/api/themerrdb')
Expand Down
4 changes: 2 additions & 2 deletions tests/functional/test_server_ui.py
Original file line number Diff line number Diff line change
Expand Up @@ -173,10 +173,10 @@ def test_scoped_dashboard_and_playback_url_do_not_mix_identical_rating_keys(clie


def test_publication_api_requires_login_and_returns_cached_status(client, monkeypatch):
from themerr import github_status
from themerr import deployment_status
status = {'updated_at': '2026-10-02T12:00:00+00:00', 'next_check': 12345, 'stale': False}
lookup = Mock(return_value=status)
monkeypatch.setattr(github_status, 'publication_status', lookup)
monkeypatch.setattr(deployment_status, 'publication_status', lookup)
with TestClient(client.app, follow_redirects=False) as anonymous:
assert anonymous.get('/api/themerrdb').status_code == 401
lookup.assert_not_called()
Expand Down
221 changes: 221 additions & 0 deletions tests/unit/themerr/test_deployment_status.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,221 @@
"""Verify published deployment timestamps, persisted caching, and failure behavior."""

# standard imports
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Event
from unittest.mock import Mock

# lib imports
import pytest
import requests

# local imports
from themerr import deployment_status, storage

PUBLICATION = '2026-10-02T12:00:00Z'
DEPLOYMENT_URL = 'https://app.lizardbyte.dev/ThemerrDB/deployment.json'
REPOSITORY_URL = 'https://github.com/LizardByte/ThemerrDB'


def response(payload=None, status=200):
value = Mock(status_code=status)
value.json.return_value = payload if payload is not None else {
'deployed_at': PUBLICATION,
'schemaVersion': 1,
'label': 'last deployment',
'message': '2026-10-02',
'color': 'brightgreen',
}
value.__enter__ = Mock(return_value=value)
value.__exit__ = Mock(return_value=False)
return value


def test_hourly_limit_survives_restart_and_server_scopes(configured, monkeypatch):
clock = Mock(return_value=10000)
get = Mock(return_value=response())
monkeypatch.setattr(deployment_status.time, 'time', clock)
monkeypatch.setattr(deployment_status.requests, 'get', get)
first = deployment_status.publication_status()
assert first == {
'updated_at': '2026-10-02T12:00:00+00:00',
'url': REPOSITORY_URL,
'checked_at': 10000,
'next_check': 13600,
'stale': False,
}
get.assert_called_once_with(
DEPLOYMENT_URL,
headers={
'Accept': 'application/json',
'Cache-Control': 'no-cache',
},
timeout=10, allow_redirects=False,
)
storage.close()
clock.return_value = 13599
with storage.server_scope('another-server'):
assert deployment_status.publication_status() == first
get.assert_called_once()
clock.return_value = 13600
assert deployment_status.publication_status()['next_check'] == 17200
assert get.call_count == 2


@pytest.mark.parametrize('failure', [
requests.ConnectionError(),
requests.Timeout(),
requests.HTTPError(),
ValueError('Invalid JSON'),
TypeError('Invalid payload'),
KeyError('deployed_at'),
])
def test_failed_check_preserves_publication_and_does_not_retry_early(configured, monkeypatch, failure):
clock = Mock(return_value=10000)
get = Mock(return_value=response())
monkeypatch.setattr(deployment_status.time, 'time', clock)
monkeypatch.setattr(deployment_status.requests, 'get', get)
first = deployment_status.publication_status()
clock.return_value = 13600
get.side_effect = failure
failed = deployment_status.publication_status()
assert failed['stale'] is True
assert failed['updated_at'] == first['updated_at']
assert failed['url'] == first['url']
assert failed['next_check'] == 17200
storage.close()
clock.return_value = 17199
assert deployment_status.publication_status() == failed
assert get.call_count == 2
get.side_effect = None
clock.return_value = 17200
assert deployment_status.publication_status()['stale'] is False
assert get.call_count == 3


@pytest.mark.parametrize('payload', [
{'deployed_at': '2026-10-02T12:00:00'},
{'deployed_at': '2026-10-02'},
{'deployed_at': 'invalid'},
{'deployed_at': ''},
{'deployed_at': None},
{'deployed_at': 123456},
{'deployed_at': {}},
{'deployed_at': []},
{},
[],
'invalid',
123456,
])
def test_invalid_response_is_not_published_or_retried(configured, monkeypatch, payload):
get = Mock(return_value=response(payload))
monkeypatch.setattr(deployment_status.requests, 'get', get)
failed = deployment_status.publication_status()
assert failed['updated_at'] is None
assert failed['stale'] is True
assert failed['url'] == REPOSITORY_URL
assert deployment_status.publication_status() == failed
get.assert_called_once()


@pytest.mark.parametrize('status', [
204,
302,
404,
500,
])
def test_unsuccessful_response_is_not_published_or_retried(configured, monkeypatch, status):
get = Mock(return_value=response(status=status))
monkeypatch.setattr(deployment_status.requests, 'get', get)
failed = deployment_status.publication_status()
assert failed['updated_at'] is None
assert failed['stale'] is True
assert failed['url'] == REPOSITORY_URL
assert deployment_status.publication_status() == failed
get.assert_called_once()


@pytest.mark.parametrize('deployed_at', [
PUBLICATION,
'2026-10-02T12:00:00+00:00',
'2026-10-02T17:30:00+05:30',
'2026-10-02T08:00:00-04:00',
])
def test_deployment_time_is_normalized_to_utc(configured, monkeypatch, deployed_at):
get = Mock(return_value=response({'deployed_at': deployed_at}))
monkeypatch.setattr(deployment_status.requests, 'get', get)
result = deployment_status.publication_status()
assert result['updated_at'] == '2026-10-02T12:00:00+00:00'
assert result['stale'] is False


def test_concurrent_page_requests_share_one_check(configured, monkeypatch):
started, release = Event(), Event()

def request(*args, **kwargs):
started.set()
assert release.wait(10)
return response()

get = Mock(side_effect=request)
monkeypatch.setattr(deployment_status.requests, 'get', get)
with ThreadPoolExecutor(max_workers=2) as executor:
first = executor.submit(deployment_status.publication_status)
assert started.wait(10)
second = executor.submit(deployment_status.publication_status)
release.set()
assert first.result(timeout=10) == second.result(timeout=10)
get.assert_called_once()


def test_interrupted_attempt_still_reserves_the_hour(configured, monkeypatch):
get = Mock(side_effect=KeyboardInterrupt())
monkeypatch.setattr(deployment_status.requests, 'get', get)
with pytest.raises(KeyboardInterrupt):
deployment_status.publication_status()
storage.close()
assert deployment_status.publication_status()['stale'] is True
get.assert_called_once()


@pytest.mark.parametrize('version', [
None,
2,
])
@pytest.mark.parametrize('failure', [
False,
True,
])
def test_legacy_cache_refreshes_once_without_losing_the_last_result(configured, monkeypatch, version, failure):
old_date = '2026-08-13T16:24:23+00:00'
deployment_status._save({
'version': version,
'updated_at': old_date,
'run_id': 999,
'next_check': 999999,
'stale': False,
})
clock = Mock(return_value=10000)
get = Mock(side_effect=requests.Timeout()) if failure else Mock(return_value=response())
monkeypatch.setattr(deployment_status.time, 'time', clock)
monkeypatch.setattr(deployment_status.requests, 'get', get)
result = deployment_status.publication_status()
assert result['updated_at'] == (old_date if failure else '2026-10-02T12:00:00+00:00')
assert result['stale'] is failure
assert result['next_check'] == 13600
assert result['url'] == REPOSITORY_URL
assert deployment_status._load()['version'] == deployment_status._CACHE_VERSION
if not failure:
assert 'run_id' not in deployment_status._load()
storage.close()
clock.return_value = 13599
assert deployment_status.publication_status() == result
get.assert_called_once()


def test_deployment_cache_uses_the_test_installation_database(configured, tmp_path, monkeypatch):
monkeypatch.setattr(deployment_status.requests, 'get', Mock(return_value=response()))
assert Path(storage.database_path()).parent == tmp_path
deployment_status.publication_status()
assert deployment_status._load()['updated_at'] == '2026-10-02T12:00:00+00:00'
Loading
Loading