152 lines
6.7 KiB
Python
152 lines
6.7 KiB
Python
import asyncio
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path, PurePosixPath
|
|
from uuid import uuid4
|
|
|
|
from websockets.asyncio.server import serve
|
|
|
|
from archive_clients.config import ClientConfig, ConnectionConfig, ServiceConfig
|
|
from archive_clients.daemon import ArchiveClientDaemon
|
|
from archive_clients.probes import FilesystemProbe
|
|
from archive_clients.services import ServiceProbe
|
|
from archive_clients.protocol import decode, encode, encode_message, new_envelope
|
|
from archive_control.v1 import client_pb2, common_pb2, control_pb2, job_pb2
|
|
|
|
|
|
class DaemonTransportTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_registration_heartbeat_and_duplicate_command(self):
|
|
observed = {}
|
|
job_id = str(uuid4())
|
|
|
|
async def control(websocket):
|
|
registration = decode(await websocket.recv())
|
|
observed["token"] = registration.register_request.shared_token
|
|
observed["root_names"] = [
|
|
item.root_name
|
|
for item in registration.register_request.capabilities.filesystems
|
|
]
|
|
observed["addresses"] = list(
|
|
registration.register_request.capabilities
|
|
.syncthing_advertised_addresses
|
|
)
|
|
observed["device_id"] = (
|
|
registration.register_request.capabilities.syncthing_device_id
|
|
)
|
|
observed["services"] = [
|
|
item.service
|
|
for item in registration.register_request.capabilities.services
|
|
]
|
|
response = new_envelope()
|
|
response.correlation_id = registration.message_id
|
|
response.register_response.status = client_pb2.REGISTRATION_STATUS_ACCEPTED
|
|
response.register_response.negotiated_version.major = 1
|
|
await websocket.send(encode(response))
|
|
heartbeat = new_envelope()
|
|
heartbeat.heartbeat.sequence = 7
|
|
await websocket.send(encode(heartbeat))
|
|
observed["heartbeat"] = decode(await websocket.recv()).heartbeat_ack.sequence
|
|
command_id = str(uuid4())
|
|
command = new_envelope()
|
|
command.command.command_id = command_id
|
|
command.command.created_at.CopyFrom(command.sent_at)
|
|
command.command.request_job_snapshot.job_ids.append(job_id)
|
|
await websocket.send(encode(command))
|
|
observed["first"] = decode(await websocket.recv()).command_ack.status
|
|
snapshot = decode(await websocket.recv())
|
|
observed["snapshot"] = snapshot.WhichOneof("payload")
|
|
observed["snapshot_job_id"] = (
|
|
snapshot.job_snapshot.job.definition.job_id
|
|
)
|
|
duplicate = new_envelope()
|
|
duplicate.command.CopyFrom(command.command)
|
|
await websocket.send(encode(duplicate))
|
|
observed["second"] = decode(await websocket.recv()).command_ack.status
|
|
observed["duplicate_snapshot"] = (
|
|
decode(await websocket.recv()).WhichOneof("payload")
|
|
)
|
|
unsupported = new_envelope()
|
|
unsupported.command.command_id = str(uuid4())
|
|
unsupported.command.created_at.CopyFrom(unsupported.sent_at)
|
|
unsupported.command.assign_job.SetInParent()
|
|
await websocket.send(encode(unsupported))
|
|
rejected = decode(await websocket.recv()).command_ack
|
|
observed["rejected"] = (rejected.status, rejected.error.code)
|
|
unsupported_duplicate = new_envelope()
|
|
unsupported_duplicate.command.CopyFrom(unsupported.command)
|
|
await websocket.send(encode(unsupported_duplicate))
|
|
observed["rejected_duplicate"] = (
|
|
decode(await websocket.recv()).command_ack.status
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
token = root / "token"
|
|
token.write_text("shared-secret", encoding="utf-8")
|
|
os.chmod(token, 0o600)
|
|
async with serve(control, "127.0.0.1", 0, ping_interval=None) as server:
|
|
port = server.sockets[0].getsockname()[1]
|
|
service = ServiceConfig(
|
|
"http://local", PurePosixPath("/api"), root,
|
|
advertised_addresses=("dynamic",),
|
|
)
|
|
config = ClientConfig(
|
|
"cache-1", "Cache 1", "cache",
|
|
f"ws://127.0.0.1:{port}", token,
|
|
root / "state.db", root / "backups", service, service,
|
|
ConnectionConfig(registration_timeout=2),
|
|
)
|
|
probe = FilesystemProbe(root, True, True, True, True, True)
|
|
service_probe = ServiceProbe(
|
|
"syncthing", common_pb2.HEALTH_STATE_HEALTHY,
|
|
datetime.now(timezone.utc), version="v2", device_id="DEVICE",
|
|
)
|
|
daemon = ArchiveClientDaemon(
|
|
config, [probe, probe], [service_probe]
|
|
)
|
|
await asyncio.to_thread(daemon.store.initialize)
|
|
definition = job_pb2.JobDefinition(
|
|
job_id=job_id,
|
|
operation=job_pb2.JOB_OPERATION_ARCHIVE,
|
|
)
|
|
definition.created_at.GetCurrentTime()
|
|
await asyncio.to_thread(
|
|
daemon.store.save_job,
|
|
job_id,
|
|
encode_message(definition),
|
|
"JOB_STATE_WAITING",
|
|
2,
|
|
3,
|
|
False,
|
|
)
|
|
await daemon._connection()
|
|
|
|
self.assertEqual(observed["token"], "shared-secret")
|
|
self.assertEqual(observed["root_names"], ["qbittorrent", "syncthing"])
|
|
self.assertEqual(observed["addresses"], ["dynamic"])
|
|
self.assertEqual(observed["device_id"], "DEVICE")
|
|
self.assertEqual(observed["services"], ["syncthing"])
|
|
self.assertEqual(observed["heartbeat"], 7)
|
|
self.assertEqual(observed["first"], control_pb2.COMMAND_ACK_STATUS_ACCEPTED)
|
|
self.assertEqual(observed["second"], control_pb2.COMMAND_ACK_STATUS_DUPLICATE)
|
|
self.assertEqual(observed["snapshot"], "job_snapshot")
|
|
self.assertEqual(observed["snapshot_job_id"], job_id)
|
|
self.assertEqual(observed["duplicate_snapshot"], "job_snapshot")
|
|
self.assertEqual(
|
|
observed["rejected"],
|
|
(
|
|
control_pb2.COMMAND_ACK_STATUS_REJECTED,
|
|
common_pb2.ERROR_CODE_UNSUPPORTED,
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
observed["rejected_duplicate"],
|
|
control_pb2.COMMAND_ACK_STATUS_REJECTED,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|