style: format the last 10 files that ruff format disagreed with

ruff.toml has declared `quote-style = "single"` and line-length 120 since the
lint job landed, and 118 of 128 files follow it. The panel backend and the
netbox configuration were written in black/prettier style instead, so a
`ruff format --check` would have failed on them from the start.

Bring them onto the style the repository already declares, which is what makes
the check adoptable at all. Formatting only: apart from quote style the diff is
multi-line expressions joined where they fit inside 120 columns.

Both suites still pass afterwards (25 pytest, and ruff check is clean).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
forustandClaude Opus 4.8 committed 2026-09-27 10:19:23 +02:00
1 parent fddd82704f
commit 0ae0df7473
10 files changed
+320 -334

No files matched your search

@@ -23,7 +23,7 @@ class FakeClient:
self.disconnected = True
async def send_code(self, _phone):
return SimpleNamespace(phone_code_hash="hash")
return SimpleNamespace(phone_code_hash='hash')
async def sign_in(self, *_args):
if self.requires_password:
@@ -38,49 +38,49 @@ class FakeClient:
return SimpleNamespace(id=1)
async def export_session_string(self):
return "exported-session"
return 'exported-session'
def phone_payload() -> PhoneStart:
return PhoneStart(
instance_id="test",
display_name="Test",
instance_id='test',
display_name='Test',
api_id=123,
api_hash="0123456789abcdef",
phone="+421900000000",
api_hash='0123456789abcdef',
phone='+421900000000',
)
@pytest.mark.asyncio
async def test_phone_code_success_closes_and_forgets_client(monkeypatch) -> None:
monkeypatch.setattr("app.auth_service.Client", FakeClient)
monkeypatch.setattr('app.auth_service.Client', FakeClient)
auth = TelegramAuthService()
flow_id = await auth.start_phone(phone_payload())
result = await auth.submit_code(flow_id, "12345")
result = await auth.submit_code(flow_id, '12345')
assert result.session_string == "exported-session"
assert result.session_string == 'exported-session'
assert flow_id not in auth.flows
assert FakeClient.instances[-1].disconnected is True
assert FakeClient.instances[-1].kwargs["in_memory"] is True
assert FakeClient.instances[-1].kwargs["no_updates"] is True
assert FakeClient.instances[-1].kwargs['in_memory'] is True
assert FakeClient.instances[-1].kwargs['no_updates'] is True
@pytest.mark.asyncio
async def test_string_session_is_validated_and_closed(monkeypatch) -> None:
monkeypatch.setattr("app.auth_service.Client", FakeClient)
monkeypatch.setattr('app.auth_service.Client', FakeClient)
auth = TelegramAuthService()
payload = StringSessionStart(
instance_id="test",
display_name="Test",
instance_id='test',
display_name='Test',
api_id=123,
api_hash="0123456789abcdef",
session_string="x" * 64,
api_hash='0123456789abcdef',
session_string='x' * 64,
)
result = await auth.validate_string_session(payload)
assert result.session_string == "exported-session"
assert result.session_string == 'exported-session'
assert FakeClient.instances[-1].disconnected is True
@@ -88,7 +88,7 @@ async def test_string_session_is_validated_and_closed(monkeypatch) -> None:
async def test_start_phone_rejects_overflow(monkeypatch) -> None:
from app.errors import PanelError
monkeypatch.setattr("app.auth_service.Client", FakeClient)
monkeypatch.setattr('app.auth_service.Client', FakeClient)
auth = TelegramAuthService(ttl_seconds=3600, max_flows=2)
await auth.start_phone(phone_payload())
@@ -98,4 +98,4 @@ async def test_start_phone_rejects_overflow(monkeypatch) -> None:
await auth.start_phone(phone_payload())
assert error.value.status_code == 429
assert "Too many pending" in error.value.detail
assert 'Too many pending' in error.value.detail
@@ -20,35 +20,35 @@ def service() -> KubernetesService:
def account() -> AccountBase:
return AccountBase(
instance_id="test-account",
display_name="Test Account",
instance_id='test-account',
display_name='Test Account',
api_id=12345,
api_hash="0123456789abcdef0123456789abcdef",
api_hash='0123456789abcdef0123456789abcdef',
)
def test_renders_managed_resources_without_leaking_credentials() -> None:
kube = service()
names = kube._resource_names("test-account")
secret = kube._secret(account(), "SESSION", names)
names = kube._resource_names('test-account')
secret = kube._secret(account(), 'SESSION', names)
pvc = kube._pvc(account(), names)
deployment = kube._deployment(account(), names)
assert secret.string_data == {
"API_ID": "12345",
"API_HASH": "0123456789abcdef0123456789abcdef",
"STRINGSESSION": "SESSION",
'API_ID': '12345',
'API_HASH': '0123456789abcdef0123456789abcdef',
'STRINGSESSION': 'SESSION',
}
assert pvc.spec.storage_class_name == "local-path-retain"
assert pvc.spec.resources.requests["storage"] == "1Gi"
assert deployment.spec.strategy.type == "Recreate"
assert pvc.spec.storage_class_name == 'local-path-retain'
assert pvc.spec.resources.requests['storage'] == '1Gi'
assert deployment.spec.strategy.type == 'Recreate'
assert deployment.spec.replicas == 1
assert deployment.spec.template.spec.automount_service_account_token is False
assert deployment.spec.template.spec.service_account_name == "userbot-runtime"
assert deployment.spec.template.spec.volumes[1].host_path.path.endswith("/Downloads")
assert deployment.spec.template.spec.service_account_name == 'userbot-runtime'
assert deployment.spec.template.spec.volumes[1].host_path.path.endswith('/Downloads')
assert deployment.spec.template.spec.containers[0].resources.requests == {
"cpu": "80m",
"memory": "512Mi",
'cpu': '80m',
'memory': '512Mi',
}
@@ -57,10 +57,10 @@ def test_collision_is_reported_before_auth() -> None:
kube.apps.read_namespaced_deployment.return_value = object()
with pytest.raises(PanelError) as error:
kube.assert_available("test-account")
kube.assert_available('test-account')
assert error.value.status_code == 409
assert "Deployment" in error.value.detail
assert 'Deployment' in error.value.detail
def test_partial_provision_rolls_back_only_created_resources() -> None:
@@ -70,11 +70,11 @@ def test_partial_provision_rolls_back_only_created_resources() -> None:
kube.core.create_namespaced_persistent_volume_claim.side_effect = ApiException(status=500)
with pytest.raises(PanelError):
kube.provision(account(), "SESSION")
kube.provision(account(), 'SESSION')
kube.core.delete_namespaced_secret.assert_called_once_with(
"userbot-test-account-credentials",
"userbot",
'userbot-test-account-credentials',
'userbot',
)
kube.core.delete_namespaced_persistent_volume_claim.assert_not_called()
kube.apps.delete_namespaced_deployment.assert_not_called()
@@ -86,18 +86,18 @@ def test_provision_conflict_reports_existing_instance() -> None:
kube.apps.create_namespaced_deployment.side_effect = ApiException(status=409)
with pytest.raises(PanelError) as error:
kube.provision(account(), "SESSION")
kube.provision(account(), 'SESSION')
assert error.value.status_code == 409
assert "already exists" in error.value.detail
assert 'already exists' in error.value.detail
# Partial resources created before the 409 must be rolled back.
kube.core.delete_namespaced_secret.assert_called_once_with(
"userbot-test-account-credentials",
"userbot",
'userbot-test-account-credentials',
'userbot',
)
kube.core.delete_namespaced_persistent_volume_claim.assert_called_once_with(
"userbot-test-account-data",
"userbot",
'userbot-test-account-data',
'userbot',
)
@@ -105,25 +105,25 @@ def test_delete_retains_pvc_unless_explicitly_requested() -> None:
kube = service()
deployment = SimpleNamespace(
metadata=SimpleNamespace(
name="userbot-test-account",
name='userbot-test-account',
annotations={
"userbot.forust.xyz/credentials-secret": "credentials",
"userbot.forust.xyz/pvc": "data",
'userbot.forust.xyz/credentials-secret': 'credentials',
'userbot.forust.xyz/pvc': 'data',
},
)
)
kube._find_deployment = Mock(return_value=("userbot", deployment))
kube._find_deployment = Mock(return_value=('userbot', deployment))
kube.delete("test-account", delete_data=False)
kube.delete('test-account', delete_data=False)
kube.apps.delete_namespaced_deployment.assert_called_once()
kube.core.delete_namespaced_secret.assert_called_once_with("credentials", "userbot")
kube.core.delete_namespaced_secret.assert_called_once_with('credentials', 'userbot')
kube.core.delete_namespaced_persistent_volume_claim.assert_not_called()
kube.delete("test-account", delete_data=True)
kube.delete('test-account', delete_data=True)
kube.core.delete_namespaced_persistent_volume_claim.assert_called_once_with(
"data",
"userbot",
'data',
'userbot',
)
@@ -131,19 +131,19 @@ def test_legacy_delete_is_blocked() -> None:
kube = service()
deployment = SimpleNamespace(
metadata=SimpleNamespace(
name="forust-userbot-deployment",
annotations={"userbot.forust.xyz/legacy": "true"},
name='forust-userbot-deployment',
annotations={'userbot.forust.xyz/legacy': 'true'},
)
)
kube._find_deployment = Mock(return_value=("default", deployment))
kube._find_deployment = Mock(return_value=('default', deployment))
with pytest.raises(PanelError) as error:
kube.delete("forust", delete_data=False)
kube.delete('forust', delete_data=False)
assert error.value.status_code == 409
def pod(*, ready: bool, phase: str = "Running", reason: str | None = None):
def pod(*, ready: bool, phase: str = 'Running', reason: str | None = None):
waiting = SimpleNamespace(reason=reason) if reason else None
state = SimpleNamespace(waiting=waiting, terminated=None)
status = SimpleNamespace(
@@ -154,25 +154,25 @@ def pod(*, ready: bool, phase: str = "Running", reason: str | None = None):
start_time=datetime.now(UTC),
)
return SimpleNamespace(
metadata=SimpleNamespace(name="pod-1", creation_timestamp=datetime.now(UTC)),
metadata=SimpleNamespace(name='pod-1', creation_timestamp=datetime.now(UTC)),
status=status,
)
def deployment(replicas: int = 1):
resources = SimpleNamespace(limits={"cpu": "300m", "memory": "1536Mi"})
container = SimpleNamespace(image="userbot:latest", resources=resources)
resources = SimpleNamespace(limits={'cpu': '300m', 'memory': '1536Mi'})
container = SimpleNamespace(image='userbot:latest', resources=resources)
template_spec = SimpleNamespace(containers=[container], volumes=[])
return SimpleNamespace(
metadata=SimpleNamespace(
name="userbot-test-account",
labels={"app.kubernetes.io/instance": "test-account"},
name='userbot-test-account',
labels={'app.kubernetes.io/instance': 'test-account'},
annotations={},
creation_timestamp=datetime.now(UTC),
),
spec=SimpleNamespace(
replicas=replicas,
selector=SimpleNamespace(match_labels={"app": "test"}),
selector=SimpleNamespace(match_labels={'app': 'test'}),
template=SimpleNamespace(spec=template_spec),
),
status=SimpleNamespace(available_replicas=1 if replicas else 0),
@@ -180,13 +180,13 @@ def deployment(replicas: int = 1):
@pytest.mark.parametrize(
("replicas", "pod_value", "expected"),
('replicas', 'pod_value', 'expected'),
[
(0, None, "stopped"),
(1, pod(ready=True), "running"),
(1, pod(ready=False, phase="Pending"), "pending"),
(1, pod(ready=False, reason="CrashLoopBackOff"), "error"),
(1, pod(ready=False, phase="Failed"), "error"),
(0, None, 'stopped'),
(1, pod(ready=True), 'running'),
(1, pod(ready=False, phase='Pending'), 'pending'),
(1, pod(ready=False, reason='CrashLoopBackOff'), 'error'),
(1, pod(ready=False, phase='Failed'), 'error'),
],
)
def test_status_classification(replicas, pod_value, expected) -> None:
@@ -195,6 +195,6 @@ def test_status_classification(replicas, pod_value, expected) -> None:
kube._pvc_storage = Mock(return_value=None)
kube._pod_metrics = Mock(return_value=(None, None))
result = kube._summarize("userbot", deployment(replicas))
result = kube._summarize('userbot', deployment(replicas))
assert result.status == expected
+10 -10
View File
@@ -6,34 +6,34 @@ from pydantic import ValidationError
@pytest.mark.parametrize(
"instance_id",
["Upper", "has_space", "-leading", "trailing-", "x" * 41],
'instance_id',
['Upper', 'has_space', '-leading', 'trailing-', 'x' * 41],
)
def test_invalid_instance_ids(instance_id: str) -> None:
with pytest.raises(ValidationError):
AccountBase(
instance_id=instance_id,
display_name="Test",
display_name='Test',
api_id=1,
api_hash="0123456789abcdef",
api_hash='0123456789abcdef',
)
def test_delete_data_defaults_to_false() -> None:
request = DeleteRequest(confirmation="test")
request = DeleteRequest(confirmation='test')
assert request.delete_data is False
@pytest.mark.asyncio
async def test_validation_response_does_not_echo_secret_input() -> None:
sensitive_value = "-".join(("very", "private", "string", "session"))
sensitive_value = '-'.join(('very', 'private', 'string', 'session'))
error = RequestValidationError(
[
{
"type": "string_too_short",
"loc": ("body", "session_string"),
"msg": "String should have at least 32 characters",
"input": sensitive_value,
'type': 'string_too_short',
'loc': ('body', 'session_string'),
'msg': 'String should have at least 32 characters',
'input': sensitive_value,
}
]
)
+16 -16
View File
@@ -6,43 +6,43 @@ from app.main import spa_fallback
@pytest.fixture
def static_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
static = tmp_path / "static"
static = tmp_path / 'static'
static.mkdir()
(static / "index.html").write_text("<html>index</html>", encoding="utf-8")
(static / "app.js").write_text("console.log('app')", encoding="utf-8")
secret = tmp_path / "secret.txt"
secret.write_text("TOP SECRET", encoding="utf-8")
monkeypatch.setattr("app.main.static_dir", static)
(static / 'index.html').write_text('<html>index</html>', encoding='utf-8')
(static / 'app.js').write_text("console.log('app')", encoding='utf-8')
secret = tmp_path / 'secret.txt'
secret.write_text('TOP SECRET', encoding='utf-8')
monkeypatch.setattr('app.main.static_dir', static)
return static
def static_index(static_dir: Path) -> Path:
return static_dir / "index.html"
return static_dir / 'index.html'
def test_returns_existing_file(static_dir: Path) -> None:
response = spa_fallback("app.js")
assert response.path == static_dir / "app.js"
response = spa_fallback('app.js')
assert response.path == static_dir / 'app.js'
def test_unknown_path_falls_back_to_index(static_dir: Path) -> None:
response = spa_fallback("does/not/exist.js")
response = spa_fallback('does/not/exist.js')
assert response.path == static_index(static_dir)
def test_traversal_does_not_leak_outside_static(static_dir: Path) -> None:
response = spa_fallback("../secret.txt")
response = spa_fallback('../secret.txt')
assert response.path == static_index(static_dir)
response = spa_fallback("%2e%2e/secret.txt")
response = spa_fallback('%2e%2e/secret.txt')
assert response.path == static_index(static_dir)
def test_symlink_outside_static_is_blocked(static_dir: Path, tmp_path: Path) -> None:
target = tmp_path / "outside.txt"
target.write_text("secret", encoding="utf-8")
link = static_dir / "leak.txt"
target = tmp_path / 'outside.txt'
target.write_text('secret', encoding='utf-8')
link = static_dir / 'leak.txt'
link.symlink_to(target)
response = spa_fallback("leak.txt")
response = spa_fallback('leak.txt')
assert response.path == static_index(static_dir)