mirror of
https://github.com/jhd3197/ServerKit.git
synced 2026-11-09 13:35:12 +09:00
Juan DenisandClaude Opus 5
9d570834aa
The `app` fixture called create_app('testing') for every one of 3173 tests,
registering 90+ blueprints each time. Measured on a fixed 199-test sample
(pytest --splits 16 --group 5 --durations=0):
setup 144.9s 89.5%
call 8.8s 5.4%
teardown 8.2s 5.0%
95% of the suite's time was preparing to test; 5% was testing. The app is now
built once per process and shared.
Same 794-test slice, before and after — identical results, 2.66x faster:
before 4 failed, 773 passed, 17 skipped 593.30s
after 4 failed, 773 passed, 17 skipped 223.11s
(the 4 are pre-existing locally: serverkit-wordpress isn't checked out here.)
Sharing one app surfaces three things a per-test app hid, all handled in the
fixture rather than patched test by test:
1. create_app() is not a pure constructor — it SEEDS the bundled flagship
extension rows as a boot side effect (test_cloudflare_extraction asserts
they are "seeded on boot"). The per-test drop_all wipes them, and a shared
app never re-boots, so every test after the first saw no flagship and the
extension routes answered 503. seed_flagship_extensions() is now re-run
after each create_all. The session fixture also boots twice on purpose:
once to create the schema, once to seed into a schema that exists.
2. Config mutation now leaks where it used to die with the app (e.g.
test_demo_deploys sets DEMO_DEPLOYS_ENABLED=False and never restores it),
so config is snapshotted and restored around every test.
3. Structure mutation cannot be undone at all — Flask has no
unregister_blueprint, and refuses setup methods once an app has served a
request. Those tests opt out via `pytestmark = pytest.mark.fresh_app` and
get a private app. Three groups needed it: 13 modules calling
register_blueprint/add_url_rule directly, 11 that install plugins (
plugin_service hot-loads their blueprints onto the live app), and
test_audit_log which adds routes with @app.route.
Tests using the `wp_extension` fixture are detected from the fixture graph
instead of being marked, since that fixture mounts blueprints on whatever app
it is handed — miss one marker and the failure lands in an unrelated test much
later in the run.
Plan 64 Phase 1. Phase 2 (savepoint rollback instead of create_all/drop_all)
is NOT included: it currently takes the slice from 4 failures to 100 failures
and 201 errors, and needs its own round.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
303 lines
9.8 KiB
Python
303 lines
9.8 KiB
Python
"""Tests for the on-demand server speed test (service tiers + API).
|
|
|
|
All network access is stubbed — the suite stays offline.
|
|
"""
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from app.services import speed_test_service as sts_module
|
|
from app.services.speed_test_service import (
|
|
LAST_RESULT_KEY,
|
|
SPEEDTEST_JOB_KIND,
|
|
SpeedTestService,
|
|
)
|
|
|
|
# Registers blueprints / url rules on the app fixture. Flask cannot unregister
|
|
# those, so these tests need a private app rather than the session-wide one
|
|
# (plan 64 Phase 1).
|
|
pytestmark = pytest.mark.fresh_app
|
|
|
|
|
|
class _FakeProc:
|
|
def __init__(self, returncode=0, stdout='', stderr=''):
|
|
self.returncode = returncode
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, body=b'', status=200):
|
|
self._body = body
|
|
self.status_code = status
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise sts_module.requests.HTTPError(f'HTTP {self.status_code}')
|
|
|
|
def iter_content(self, chunk_size=65536):
|
|
for i in range(0, len(self._body), chunk_size):
|
|
yield self._body[i:i + chunk_size]
|
|
|
|
|
|
class _FakeSocket:
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def no_cli(monkeypatch):
|
|
"""Pretend no speedtest binary exists — forces the pure-Python tier."""
|
|
monkeypatch.setattr(sts_module, 'is_command_available', lambda cmd: False)
|
|
|
|
|
|
# ── fallback tier ────────────────────────────────────────────────────────────
|
|
|
|
def test_fallback_full_success(no_cli, monkeypatch):
|
|
monkeypatch.setattr(
|
|
sts_module.socket, 'create_connection',
|
|
lambda addr, timeout=None: _FakeSocket(),
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.requests, 'get',
|
|
lambda url, timeout=None, stream=False: _FakeResponse(b'x' * 1_000_000),
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.requests, 'post',
|
|
lambda url, data=None, timeout=None: _FakeResponse(),
|
|
)
|
|
|
|
result = SpeedTestService.run_test()
|
|
|
|
assert result['success'] is True
|
|
assert result['method'] == 'fallback'
|
|
assert result['download_mbps'] > 0
|
|
assert result['upload_mbps'] > 0
|
|
assert result['latency_ms'] is not None and result['latency_ms'] >= 0
|
|
assert result['tested_at']
|
|
|
|
|
|
def test_fallback_upload_failure_is_best_effort(no_cli, monkeypatch):
|
|
monkeypatch.setattr(
|
|
sts_module.socket, 'create_connection',
|
|
lambda addr, timeout=None: _FakeSocket(),
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.requests, 'get',
|
|
lambda url, timeout=None, stream=False: _FakeResponse(b'x' * 500_000),
|
|
)
|
|
|
|
def _post_fails(url, data=None, timeout=None):
|
|
raise sts_module.requests.ConnectionError('upload endpoint unreachable')
|
|
|
|
monkeypatch.setattr(sts_module.requests, 'post', _post_fails)
|
|
|
|
result = SpeedTestService.run_test()
|
|
|
|
assert result['success'] is True
|
|
assert result['download_mbps'] > 0
|
|
assert result['upload_mbps'] is None # best-effort
|
|
|
|
|
|
def test_fallback_total_failure_returns_error(no_cli, monkeypatch):
|
|
def _connect_fails(addr, timeout=None):
|
|
raise OSError('network down')
|
|
|
|
def _get_fails(url, timeout=None, stream=False):
|
|
raise sts_module.requests.ConnectionError('no route')
|
|
|
|
def _post_fails(url, data=None, timeout=None):
|
|
raise sts_module.requests.ConnectionError('no route')
|
|
|
|
monkeypatch.setattr(sts_module.socket, 'create_connection', _connect_fails)
|
|
monkeypatch.setattr(sts_module.requests, 'get', _get_fails)
|
|
monkeypatch.setattr(sts_module.requests, 'post', _post_fails)
|
|
|
|
result = SpeedTestService.run_test()
|
|
|
|
assert result['success'] is False
|
|
assert 'error' in result
|
|
|
|
|
|
# ── CLI tier ─────────────────────────────────────────────────────────────────
|
|
|
|
def test_cli_ookla_json_parsed(monkeypatch):
|
|
monkeypatch.setattr(
|
|
sts_module, 'is_command_available', lambda cmd: cmd == 'speedtest'
|
|
)
|
|
ookla = json.dumps({
|
|
'download': {'bandwidth': 12_500_000}, # bytes/s -> 100 Mbps
|
|
'upload': {'bandwidth': 2_500_000}, # bytes/s -> 20 Mbps
|
|
'ping': {'latency': 8.42},
|
|
})
|
|
monkeypatch.setattr(
|
|
sts_module.subprocess, 'run',
|
|
lambda cmd, capture_output=True, text=True, timeout=None: _FakeProc(0, ookla),
|
|
)
|
|
|
|
result = SpeedTestService.run_test()
|
|
|
|
assert result['success'] is True
|
|
assert result['method'] == 'cli'
|
|
assert result['download_mbps'] == 100.0
|
|
assert result['upload_mbps'] == 20.0
|
|
assert result['latency_ms'] == 8.42
|
|
|
|
|
|
def test_cli_python_speedtest_cli_json_parsed(monkeypatch):
|
|
monkeypatch.setattr(
|
|
sts_module, 'is_command_available', lambda cmd: cmd == 'speedtest-cli'
|
|
)
|
|
legacy = json.dumps({
|
|
'download': 50_000_000.0, # bits/s -> 50 Mbps
|
|
'upload': 10_000_000.0, # bits/s -> 10 Mbps
|
|
'ping': 22.5,
|
|
})
|
|
monkeypatch.setattr(
|
|
sts_module.subprocess, 'run',
|
|
lambda cmd, capture_output=True, text=True, timeout=None: _FakeProc(0, legacy),
|
|
)
|
|
|
|
result = SpeedTestService.run_test()
|
|
|
|
assert result['success'] is True
|
|
assert result['method'] == 'cli'
|
|
assert result['download_mbps'] == 50.0
|
|
assert result['upload_mbps'] == 10.0
|
|
assert result['latency_ms'] == 22.5
|
|
|
|
|
|
def test_cli_failure_falls_back_to_python_tier(monkeypatch):
|
|
monkeypatch.setattr(
|
|
sts_module, 'is_command_available', lambda cmd: cmd == 'speedtest'
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.subprocess, 'run',
|
|
lambda cmd, capture_output=True, text=True, timeout=None: _FakeProc(1, '', 'boom'),
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.socket, 'create_connection',
|
|
lambda addr, timeout=None: _FakeSocket(),
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.requests, 'get',
|
|
lambda url, timeout=None, stream=False: _FakeResponse(b'x' * 100_000),
|
|
)
|
|
monkeypatch.setattr(
|
|
sts_module.requests, 'post',
|
|
lambda url, data=None, timeout=None: _FakeResponse(),
|
|
)
|
|
|
|
result = SpeedTestService.run_test()
|
|
|
|
assert result['success'] is True
|
|
assert result['method'] == 'fallback'
|
|
|
|
|
|
# ── job handler ──────────────────────────────────────────────────────────────
|
|
|
|
def test_job_handler_stores_last_result(app, monkeypatch):
|
|
from app.services.settings_service import SettingsService
|
|
|
|
fixed = {
|
|
'success': True, 'method': 'fallback', 'download_mbps': 42.0,
|
|
'upload_mbps': 7.5, 'latency_ms': 12.0, 'tested_at': '2026-07-03T00:00:00Z',
|
|
}
|
|
monkeypatch.setattr(SpeedTestService, 'run_test', classmethod(lambda cls: fixed))
|
|
|
|
class _JobStub:
|
|
def get_payload(self):
|
|
return {}
|
|
|
|
result = SpeedTestService.run_speed_test_job(_JobStub())
|
|
|
|
assert result == fixed
|
|
stored = SettingsService.get(LAST_RESULT_KEY)
|
|
assert json.loads(stored) == fixed
|
|
|
|
|
|
def test_register_jobs_registers_handler(app):
|
|
from app.jobs import registry
|
|
|
|
SpeedTestService.register_jobs()
|
|
assert registry.get(SPEEDTEST_JOB_KIND) is not None
|
|
|
|
|
|
# ── API ──────────────────────────────────────────────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def st_client(app):
|
|
"""Test client with the speedtest blueprint mounted (registration in
|
|
app/__init__.py is wired separately)."""
|
|
from app.api.speed_test import speedtest_bp
|
|
if 'speedtest' not in app.blueprints:
|
|
app.register_blueprint(speedtest_bp, url_prefix='/api/v1/speedtest')
|
|
return app.test_client()
|
|
|
|
|
|
def test_get_returns_last_result_and_running_flag(st_client, auth_headers, app):
|
|
from app.services.settings_service import SettingsService
|
|
|
|
fixed = {'success': True, 'method': 'cli', 'download_mbps': 90.0,
|
|
'upload_mbps': 30.0, 'latency_ms': 5.0,
|
|
'tested_at': '2026-07-03T01:00:00Z'}
|
|
SettingsService.set(LAST_RESULT_KEY, json.dumps(fixed))
|
|
|
|
resp = st_client.get('/api/v1/speedtest', headers=auth_headers)
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['last_result'] == fixed
|
|
assert data['running'] is False
|
|
assert data['job'] is None
|
|
|
|
|
|
def test_get_with_no_result_yet(st_client, auth_headers):
|
|
resp = st_client.get('/api/v1/speedtest', headers=auth_headers)
|
|
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data['last_result'] is None
|
|
assert data['running'] is False
|
|
|
|
|
|
def test_run_enqueues_job(st_client, auth_headers, monkeypatch):
|
|
from app.jobs.service import JobService
|
|
|
|
enqueued = []
|
|
|
|
class _FakeJob:
|
|
id = 'job-123'
|
|
|
|
def _fake_enqueue(kind, payload=None, max_attempts=3, **kwargs):
|
|
enqueued.append((kind, payload, max_attempts))
|
|
return _FakeJob()
|
|
|
|
monkeypatch.setattr(JobService, 'enqueue', staticmethod(_fake_enqueue))
|
|
|
|
resp = st_client.post('/api/v1/speedtest/run', headers=auth_headers)
|
|
|
|
assert resp.status_code == 202
|
|
assert resp.get_json() == {'job_id': 'job-123'}
|
|
assert enqueued == [(SPEEDTEST_JOB_KIND, {}, 1)]
|
|
|
|
|
|
def test_run_rejects_when_already_running(st_client, auth_headers, app):
|
|
from app import db
|
|
from app.jobs.models import Job
|
|
|
|
job = Job(kind=SPEEDTEST_JOB_KIND, status=Job.STATUS_RUNNING)
|
|
job.set_payload({})
|
|
db.session.add(job)
|
|
db.session.commit()
|
|
|
|
resp = st_client.post('/api/v1/speedtest/run', headers=auth_headers)
|
|
|
|
assert resp.status_code == 409
|
|
assert 'error' in resp.get_json()
|
|
|
|
|
|
def test_endpoints_require_auth(st_client):
|
|
assert st_client.get('/api/v1/speedtest').status_code == 401
|
|
assert st_client.post('/api/v1/speedtest/run').status_code == 401
|