diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml new file mode 100644 index 00000000..e793f3b1 --- /dev/null +++ b/.github/workflows/python-tests.yml @@ -0,0 +1,49 @@ +name: python-tests + +on: + pull_request: + branches: + - main + push: + branches: + - main + +permissions: + contents: read + +jobs: + test: + name: Python (${{ matrix.python-version }}) + runs-on: ubuntu-latest + timeout-minutes: 10 + strategy: + fail-fast: false + matrix: + python-version: ["3.11", "3.12", "3.13"] + + steps: + - name: Check out code + uses: actions/checkout@v7 + + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v6 + with: + python-version: ${{ matrix.python-version }} + + - name: Install buf + uses: bufbuild/buf-setup-action@v1 + with: + version: "1.71.0" + + - name: Generate protobuf Python code (including buf/validate) + run: | + cd proto + buf generate + + - name: Install test dependencies + run: | + pip install connectrpc>=0.10.0 protovalidate werkzeug pytest + + - name: Run Python client tests + run: | + PYTHONPATH="$GITHUB_WORKSPACE/python" python -m pytest python/metalstack/client/tests/ -v diff --git a/python/metalstack/client/tests/__init__.py b/python/metalstack/client/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/python/metalstack/client/tests/conftest.py b/python/metalstack/client/tests/conftest.py new file mode 100644 index 00000000..8cf48df9 --- /dev/null +++ b/python/metalstack/client/tests/conftest.py @@ -0,0 +1,172 @@ +from collections.abc import Iterator +from typing import Any + +import pyqwest +import pytest +from connectrpc.request import RequestContext +from pyqwest.testing import WSGITransport + +from metalstack.api.v2 import ip_pb2 +from metalstack.api.v2 import version_pb2 +from metalstack.api.v2.ip_connect import ( + IPServiceWSGIApplication, +) +from metalstack.api.v2.version_connect import ( + VersionServiceWSGIApplication, +) +from metalstack.client.client import Client +from metalstack.infra.v2 import bmc_pb2 +from metalstack.infra.v2 import component_pb2 as infra_component_pb2 +from metalstack.infra.v2.bmc_connect import ( + BMCServiceWSGIApplication, +) +from metalstack.infra.v2.component_connect import ( + ComponentServiceWSGIApplication, +) + + +class MockVersionService: + def __init__(self, version: str = "1.0"): + self._version = version + + def get( + self, request: version_pb2.VersionServiceGetRequest, ctx: RequestContext + ) -> version_pb2.VersionServiceGetResponse: + return version_pb2.VersionServiceGetResponse( + version=version_pb2.Version(version=self._version) + ) + + +class MockIPService: + def __init__(self): + self.received_auth: str | None = None + + def get( + self, request: ip_pb2.IPServiceGetRequest, ctx: RequestContext + ) -> ip_pb2.IPServiceGetResponse: + self.received_auth = ctx.request_headers().get("authorization") + return ip_pb2.IPServiceGetResponse( + ip=ip_pb2.IP(ip=request.ip, project=request.project) + ) + + def create( + self, request: ip_pb2.IPServiceCreateRequest, ctx: RequestContext + ) -> ip_pb2.IPServiceCreateResponse: + return ip_pb2.IPServiceCreateResponse() + + def update( + self, request: ip_pb2.IPServiceUpdateRequest, ctx: RequestContext + ) -> ip_pb2.IPServiceUpdateResponse: + return ip_pb2.IPServiceUpdateResponse() + + def list( + self, request: ip_pb2.IPServiceListRequest, ctx: RequestContext + ) -> ip_pb2.IPServiceListResponse: + return ip_pb2.IPServiceListResponse() + + def delete( + self, request: ip_pb2.IPServiceDeleteRequest, ctx: RequestContext + ) -> ip_pb2.IPServiceDeleteResponse: + return ip_pb2.IPServiceDeleteResponse() + + +class MockBMCService: + def __init__(self): + self.received_auth: str | None = None + + def update_b_m_c_info( + self, request: bmc_pb2.UpdateBMCInfoRequest, ctx: RequestContext + ) -> bmc_pb2.UpdateBMCInfoResponse: + self.received_auth = ctx.request_headers().get("authorization") + return bmc_pb2.UpdateBMCInfoResponse() + + def wait_for_b_m_c_command( + self, request: bmc_pb2.WaitForBMCCommandRequest, ctx: RequestContext + ) -> Iterator[bmc_pb2.WaitForBMCCommandResponse]: + self.received_auth = ctx.request_headers().get("authorization") + yield bmc_pb2.WaitForBMCCommandResponse(uuid="test-uuid", command_id="cmd-1") + + def b_m_c_command_done( + self, request: bmc_pb2.BMCCommandDoneRequest, ctx: RequestContext + ) -> bmc_pb2.BMCCommandDoneResponse: + return bmc_pb2.BMCCommandDoneResponse() + + +class MockComponentService: + def __init__(self): + self.pings: list[infra_component_pb2.ComponentServicePingRequest] = [] + + def ping( + self, + request: infra_component_pb2.ComponentServicePingRequest, + ctx: RequestContext, + ) -> infra_component_pb2.ComponentServicePingResponse: + self.pings.append( + infra_component_pb2.ComponentServicePingRequest( + type=request.type, + identifier=request.identifier, + started_at=request.started_at, + interval=request.interval, + version=request.version, + ) + ) + return infra_component_pb2.ComponentServicePingResponse() + + +def _build_combined_wsgi_app(services: dict[str, Any]): + """Build a combined WSGI app that dispatches by path prefix.""" + + def application(environ, start_response): + path = environ.get("PATH_INFO", "/") + for prefix, app in services.items(): + if path.startswith(prefix): + return app(environ, start_response) + from werkzeug.exceptions import NotFound + + return NotFound()(environ, start_response) + + return application + + +@pytest.fixture +def mock_version_service(): + return MockVersionService() + + +@pytest.fixture +def mock_ip_service(): + return MockIPService() + + +@pytest.fixture +def mock_bmc_service(): + return MockBMCService() + + +@pytest.fixture +def mock_component_service(): + return MockComponentService() + + +@pytest.fixture +def test_client( + mock_version_service, + mock_ip_service, + mock_bmc_service, + mock_component_service, +): + """Create a Client with WSGI-backed transport for testing.""" + services = { + "/metalstack.api.v2.VersionService": VersionServiceWSGIApplication( + mock_version_service + ), + "/metalstack.api.v2.IPService": IPServiceWSGIApplication(mock_ip_service), + "/metalstack.infra.v2.BMCService": BMCServiceWSGIApplication(mock_bmc_service), + "/metalstack.infra.v2.ComponentService": ComponentServiceWSGIApplication( + mock_component_service + ), + } + transport = WSGITransport(_build_combined_wsgi_app(services)) + c = Client(baseurl="http://test", timeout=10) + c._client = pyqwest.SyncClient(transport=transport) + return c diff --git a/python/metalstack/client/tests/test_client.py b/python/metalstack/client/tests/test_client.py new file mode 100644 index 00000000..57e3f8f2 --- /dev/null +++ b/python/metalstack/client/tests/test_client.py @@ -0,0 +1,122 @@ +"""Tests mirroring go/client/client_test.go. + +Test_Client -> test_client_apiv2_version_get +Test_ClientInterceptors -> test_client_auth_header_unary, test_client_auth_header_streaming +""" + +import pyqwest +import pytest +from pyqwest.testing import WSGITransport + +from metalstack.api.v2 import ip_pb2 +from metalstack.api.v2 import version_pb2 +from metalstack.api.v2.ip_connect import ( + IPServiceClientSync, + IPServiceWSGIApplication, +) +from metalstack.api.v2.version_connect import ( + VersionServiceClientSync, + VersionServiceWSGIApplication, +) +from metalstack.client.client import Client +from metalstack.infra.v2 import bmc_pb2 +from metalstack.infra.v2.bmc_connect import ( + BMCServiceClientSync, + BMCServiceWSGIApplication, +) +from metalstack.infra.v2 import component_pb2 as infra_component_pb2 +from metalstack.infra.v2.component_connect import ( + ComponentServiceClientSync, + ComponentServiceWSGIApplication, +) + +from .conftest import ( + MockBMCService, + MockComponentService, + MockIPService, + MockVersionService, + _build_combined_wsgi_app, +) + + +class TestClient: + """Mirrors Test_Client from client_test.go. + + Tests client instantiation and basic RPC calls through the generated Client wrapper. + """ + + def test_client_creates_service_groups(self, test_client): + """Client exposes adminv2, apiv2, infrav2 service groups.""" + assert test_client.apiv2() is not None + assert test_client.infrav2() is not None + + def test_client_apiv2_version_get(self, test_client, mock_version_service): + """Version.Get returns the expected version through the Client wrapper.""" + resp = test_client.apiv2().version().get( + request=version_pb2.VersionServiceGetRequest() + ) + assert resp is not None + assert resp.version.version == "1.0" + + def test_client_apiv2_version_get_custom_version(self, mock_version_service): + """Client can be configured with a custom version response.""" + mock_version_service._version = "2.5.3" + vs = MockVersionService(version="2.5.3") + ips = MockIPService() + bmc = MockBMCService() + cs = MockComponentService() + + services = { + "/metalstack.api.v2.VersionService": VersionServiceWSGIApplication(vs), + "/metalstack.api.v2.IPService": IPServiceWSGIApplication(ips), + "/metalstack.infra.v2.BMCService": BMCServiceWSGIApplication(bmc), + "/metalstack.infra.v2.ComponentService": ComponentServiceWSGIApplication(cs), + } + transport = WSGITransport(_build_combined_wsgi_app(services)) + c = Client(baseurl="http://test", timeout=10) + c._client = pyqwest.SyncClient(transport=transport) + + resp = c.apiv2().version().get( + request=version_pb2.VersionServiceGetRequest() + ) + assert resp.version.version == "2.5.3" + + +class TestClientInterceptors: + """Mirrors Test_ClientInterceptors from client_test.go. + + Tests that auth headers are correctly passed for both unary and streaming calls. + """ + + def test_client_auth_header_unary(self, test_client, mock_bmc_service): + """Unary call has Authorization header set.""" + token = "test-bearer-token" + resp = test_client.infrav2().bmc().update_b_m_c_info( + request=bmc_pb2.UpdateBMCInfoRequest(), + headers={"authorization": f"Bearer {token}"}, + ) + assert resp is not None + assert mock_bmc_service.received_auth == f"Bearer {token}" + + def test_client_auth_header_streaming(self, test_client, mock_bmc_service): + """Streaming call has Authorization header set.""" + token = "test-stream-token" + stream = test_client.infrav2().bmc().wait_for_b_m_c_command( + request=bmc_pb2.WaitForBMCCommandRequest(), + headers={"authorization": f"Bearer {token}"}, + ) + assert stream is not None + for cmd in stream: + assert cmd.uuid == "test-uuid" + assert mock_bmc_service.received_auth == f"Bearer {token}" + + def test_client_auth_header_ip_get(self, test_client, mock_ip_service): + """IP Get call has Authorization header set.""" + token = "ip-service-token" + resp = test_client.apiv2().ip().get( + request=ip_pb2.IPServiceGetRequest(ip="10.0.0.1"), + headers={"authorization": f"Bearer {token}"}, + ) + assert resp is not None + assert resp.ip.ip == "10.0.0.1" + assert mock_ip_service.received_auth == f"Bearer {token}" diff --git a/python/metalstack/client/tests/test_interceptor.py b/python/metalstack/client/tests/test_interceptor.py new file mode 100644 index 00000000..e381a36c --- /dev/null +++ b/python/metalstack/client/tests/test_interceptor.py @@ -0,0 +1,173 @@ +"""Tests mirroring go/client/test_interceptor_test.go. + +Tests the interceptor pattern using connectrpc's MetadataInterceptorSync, +which is the Python equivalent of Go's TestClientInterceptor. +""" + +import pyqwest +import pytest +from connectrpc.interceptor import MetadataInterceptorSync +from connectrpc.request import RequestContext +from pyqwest.testing import WSGITransport + +from metalstack.api.v2 import ip_pb2 +from metalstack.api.v2.ip_connect import ( + IPServiceClientSync, + IPServiceWSGIApplication, +) +from metalstack.client.client import Client + +from .conftest import MockIPService, _build_combined_wsgi_app + + +class AuthInterceptor: + """Client-side interceptor that injects an Authorization header. + + Mirrors the authInterceptor from go/client/client-interceptors.go. + """ + + def __init__(self, token: str): + self.token = token + + def on_start_sync(self, ctx: RequestContext) -> None: + ctx.request_headers().add("authorization", f"Bearer {self.token}") + + def on_end_sync(self, token, ctx: RequestContext, error) -> None: + pass + + +class RequestRecordingInterceptor: + """Interceptor that records the received request for assertion. + + Mirrors the TestClientInterceptor from go/client/test_interceptor.go. + """ + + def __init__(self): + self.calls: list[tuple[str, str]] = [] + + def on_start_sync(self, ctx: RequestContext) -> None: + method = ctx.method() + self.calls.append( + (method.service_name, method.name) + ) + + def on_end_sync(self, token, ctx: RequestContext, error) -> None: + pass + + +class TestInterceptor: + """Mirrors TestInterceptor from test_interceptor_test.go. + + Tests that an interceptor can intercept a call and verify request/response flow. + """ + + def test_interceptor_receives_request(self, mock_ip_service): + """Interceptor can observe the service and method being called.""" + ips = MockIPService() + from metalstack.infra.v2.bmc_connect import BMCServiceWSGIApplication + from metalstack.infra.v2.component_connect import ComponentServiceWSGIApplication + from metalstack.api.v2.version_connect import VersionServiceWSGIApplication + from metalstack.infra.v2 import bmc_pb2 + from metalstack.infra.v2 import component_pb2 as infra_component_pb2 + + class NoopBMC: + def update_b_m_c_info(self, request, ctx): + from metalstack.infra.v2.bmc_pb2 import UpdateBMCInfoResponse + return UpdateBMCInfoResponse() + def wait_for_b_m_c_command(self, request, ctx): + from metalstack.infra.v2.bmc_pb2 import WaitForBMCCommandResponse + yield WaitForBMCCommandResponse() + def b_m_c_command_done(self, request, ctx): + from metalstack.infra.v2.bmc_pb2 import BMCCommandDoneResponse + return BMCCommandDoneResponse() + + class NoopComponent: + def ping(self, request, ctx): + from metalstack.infra.v2.component_pb2 import ComponentServicePingResponse + return ComponentServicePingResponse() + + class NoopVersion: + def get(self, request, ctx): + from metalstack.api.v2.version_pb2 import VersionServiceGetResponse, Version + return VersionServiceGetResponse(version=Version(version="0.0")) + + services = { + "/metalstack.api.v2.VersionService": VersionServiceWSGIApplication(NoopVersion()), + "/metalstack.api.v2.IPService": IPServiceWSGIApplication(ips), + "/metalstack.infra.v2.BMCService": BMCServiceWSGIApplication(NoopBMC()), + "/metalstack.infra.v2.ComponentService": ComponentServiceWSGIApplication(NoopComponent()), + } + transport = WSGITransport(_build_combined_wsgi_app(services)) + + interceptor = RequestRecordingInterceptor() + + svc = IPServiceClientSync( + "http://test", + http_client=pyqwest.SyncClient(transport=transport), + interceptors=[interceptor], + ) + + resp = svc.get( + request=ip_pb2.IPServiceGetRequest(ip="1.2.3.4"), + ) + + assert resp.ip.ip == "1.2.3.4" + assert len(interceptor.calls) == 1 + assert interceptor.calls[0] == ("metalstack.api.v2.IPService", "Get") + + def test_auth_interceptor_injects_token(self): + """Auth interceptor injects Authorization header that the server receives.""" + ips = MockIPService() + from metalstack.infra.v2.bmc_connect import BMCServiceWSGIApplication + from metalstack.infra.v2.component_connect import ComponentServiceWSGIApplication + from metalstack.api.v2.version_connect import VersionServiceWSGIApplication + + class NoopBMC: + def update_b_m_c_info(self, request, ctx): + from metalstack.infra.v2.bmc_pb2 import UpdateBMCInfoResponse + return UpdateBMCInfoResponse() + def wait_for_b_m_c_command(self, request, ctx): + from metalstack.infra.v2.bmc_pb2 import WaitForBMCCommandResponse + yield WaitForBMCCommandResponse() + def b_m_c_command_done(self, request, ctx): + from metalstack.infra.v2.bmc_pb2 import BMCCommandDoneResponse + return BMCCommandDoneResponse() + + class NoopComponent: + def ping(self, request, ctx): + from metalstack.infra.v2.component_pb2 import ComponentServicePingResponse + return ComponentServicePingResponse() + + class NoopVersion: + def get(self, request, ctx): + from metalstack.api.v2.version_pb2 import VersionServiceGetResponse, Version + return VersionServiceGetResponse(version=Version(version="0.0")) + + services = { + "/metalstack.api.v2.VersionService": VersionServiceWSGIApplication(NoopVersion()), + "/metalstack.api.v2.IPService": IPServiceWSGIApplication(ips), + "/metalstack.infra.v2.BMCService": BMCServiceWSGIApplication(NoopBMC()), + "/metalstack.infra.v2.ComponentService": ComponentServiceWSGIApplication(NoopComponent()), + } + transport = WSGITransport(_build_combined_wsgi_app(services)) + + svc = IPServiceClientSync( + "http://test", + http_client=pyqwest.SyncClient(transport=transport), + interceptors=[AuthInterceptor("my-secret-token")], + ) + + resp = svc.get(request=ip_pb2.IPServiceGetRequest(ip="1.2.3.4")) + + assert resp.ip.ip == "1.2.3.4" + assert ips.received_auth == "Bearer my-secret-token" + + def test_interceptor_on_client_wrapper(self, test_client, mock_ip_service): + """Client wrapper passes headers through to the underlying service.""" + token = "wrapper-test-token" + resp = test_client.apiv2().ip().get( + request=ip_pb2.IPServiceGetRequest(ip="5.6.7.8"), + headers={"authorization": f"Bearer {token}"}, + ) + assert resp.ip.ip == "5.6.7.8" + assert mock_ip_service.received_auth == f"Bearer {token}" diff --git a/python/metalstack/client/tests/test_ping.py b/python/metalstack/client/tests/test_ping.py new file mode 100644 index 00000000..4985ede4 --- /dev/null +++ b/python/metalstack/client/tests/test_ping.py @@ -0,0 +1,103 @@ +"""Tests mirroring go/client/ping_test.go. + +Tests the ComponentService.Ping RPC which is used by the Go client's Ping() method. +The Python client doesn't have a built-in Ping() method, but we test the underlying +ComponentService call that would be used to implement one. +""" + +import threading +import time +from datetime import timedelta + +import pytest +from google.protobuf import duration_pb2, timestamp_pb2 + +from metalstack.api.v2 import component_pb2 as api_component_pb2 +from metalstack.api.v2 import version_pb2 +from metalstack.infra.v2 import component_pb2 as infra_component_pb2 + + +class TestPing: + """Mirrors Test_Ping from ping_test.go. + + Verifies that ComponentServicePingRequest is correctly constructed and sent, + and that the mock service receives the expected fields. + """ + + def test_ping_request_fields(self, test_client, mock_component_service): + """Ping request contains correct component type, identifier, started_at, interval, version.""" + started_at = timestamp_pb2.Timestamp(seconds=1000) + interval = duration_pb2.Duration(seconds=5) + version = version_pb2.Version(version="1.0.0", git_sha1="abc123") + + resp = test_client.infrav2().component().ping( + request=infra_component_pb2.ComponentServicePingRequest( + type=api_component_pb2.COMPONENT_TYPE_METAL_CORE, + identifier="server01", + started_at=started_at, + interval=interval, + version=version, + ), + headers={"authorization": "Bearer test-token"}, + ) + + assert resp is not None + assert len(mock_component_service.pings) == 1 + + ping = mock_component_service.pings[0] + assert ping.type == api_component_pb2.COMPONENT_TYPE_METAL_CORE + assert ping.identifier == "server01" + assert ping.started_at.seconds == 1000 + assert ping.interval.seconds == 5 + assert ping.version.version == "1.0.0" + assert ping.version.git_sha1 == "abc123" + + def test_ping_multiple_requests(self, test_client, mock_component_service): + """Multiple ping calls are recorded correctly.""" + started_at = timestamp_pb2.Timestamp(seconds=2000) + interval = duration_pb2.Duration(seconds=10) + + for _ in range(3): + test_client.infrav2().component().ping( + request=infra_component_pb2.ComponentServicePingRequest( + type=api_component_pb2.COMPONENT_TYPE_METAL_CORE, + identifier="server02", + started_at=started_at, + interval=interval, + ), + headers={"authorization": "Bearer test-token"}, + ) + + assert len(mock_component_service.pings) == 3 + for ping in mock_component_service.pings: + assert ping.identifier == "server02" + assert ping.type == api_component_pb2.COMPONENT_TYPE_METAL_CORE + assert ping.started_at.seconds == 2000 + assert ping.interval.seconds == 10 + + def test_ping_background_simulation(self, test_client, mock_component_service): + """Simulate background pinging from a thread, similar to Go's ticker goroutine.""" + started_at = timestamp_pb2.Timestamp(seconds=3000) + interval = duration_pb2.Duration(seconds=1) + ping_count = 5 + + def do_pings(): + for _ in range(ping_count): + test_client.infrav2().component().ping( + request=infra_component_pb2.ComponentServicePingRequest( + type=api_component_pb2.COMPONENT_TYPE_METAL_CORE, + identifier="background-worker", + started_at=started_at, + interval=interval, + ), + headers={"authorization": "Bearer test-token"}, + ) + time.sleep(0.05) + + thread = threading.Thread(target=do_pings) + thread.start() + thread.join() + + assert len(mock_component_service.pings) == ping_count + for ping in mock_component_service.pings: + assert ping.identifier == "background-worker" diff --git a/python/metalstack/client/tests/test_token_persistent.py b/python/metalstack/client/tests/test_token_persistent.py new file mode 100644 index 00000000..2b84d867 --- /dev/null +++ b/python/metalstack/client/tests/test_token_persistent.py @@ -0,0 +1,91 @@ +"""Tests mirroring go/client/token-persistent-funcs_test.go. + +Tests filesystem-based token persistence, which is the Python equivalent of +Go's NewFilesystemTokenPersister. +""" + +import os +import tempfile +from pathlib import Path + +import pytest + + +class FilesystemTokenPersister: + """Persists a token to a file on disk. + + Mirrors the filesystem struct and persist method from + go/client/token-persistent-funcs.go. + """ + + def __init__(self, token_path: str): + self._token_path = token_path + + def persist(self, token: str) -> None: + Path(self._token_path).write_text(token) + + +def new_filesystem_token_persister(token_file: str) -> FilesystemTokenPersister: + """Create a new filesystem token persister. + + Mirrors NewFilesystemTokenPersister from go/client/token-persistent-funcs.go. + """ + path = Path(token_file) + if not path.exists(): + raise FileNotFoundError(f"token file {token_file} does not exist") + if os.access(token_file, os.W_OK) is False: + raise ValueError(f"tokenfile {token_file} is not writable") + return FilesystemTokenPersister(token_file) + + +class TestFilesystemTokenPersister: + """Mirrors TestNewFilesystemTokenPersiter from token-persistent-funcs_test.go.""" + + def test_persist_and_read_back(self, tmp_path): + """Write a token to a file and read it back.""" + token_path = str(tmp_path / "token") + Path(token_path).touch(mode=0o600) + + persister = new_filesystem_token_persister(token_path) + token = "a-token" + + persister.persist(token) + + content = Path(token_path).read_text() + assert content == token + + def test_persister_requires_existing_file(self, tmp_path): + """Persister raises FileNotFoundError for non-existent file.""" + token_path = str(tmp_path / "nonexistent") + with pytest.raises(FileNotFoundError): + new_filesystem_token_persister(token_path) + + def test_persister_requires_writable_file(self, tmp_path): + """Persister raises ValueError for read-only file.""" + token_path = str(tmp_path / "readonly") + Path(token_path).touch(mode=0o444) + + with pytest.raises(ValueError, match="not writable"): + new_filesystem_token_persister(token_path) + + def test_overwrite_existing_token(self, tmp_path): + """Persisting a new token overwrites the previous one.""" + token_path = str(tmp_path / "token") + Path(token_path).write_text("old-token") + + persister = new_filesystem_token_persister(token_path) + persister.persist("new-token") + + content = Path(token_path).read_text() + assert content == "new-token" + + def test_persist_empty_token(self, tmp_path): + """Persisting an empty token writes an empty file.""" + token_path = str(tmp_path / "token") + Path(token_path).touch(mode=0o600) + + persister = new_filesystem_token_persister(token_path) + persister.persist("") + + content = Path(token_path).read_text() + assert content == ""