anki-connect/tests/test_server.py

150 lines
4.8 KiB
Python
Raw Normal View History

import json
import multiprocessing
import time
import urllib.error
import urllib.request
from contextlib import contextmanager
from dataclasses import dataclass
from functools import partial
import pytest
from pytest_anki._launch import anki_running # noqa
from pytest_anki._util import find_free_port # noqa
from plugin import AnkiConnect
from tests.conftest import wait_until, \
empty_anki_session_started, \
anki_connect_config_loaded, \
profile_loaded
@contextmanager
def function_running_in_a_process(context, function):
process = context.Process(target=function)
process.start()
try:
yield process
finally:
process.join()
# todo stop the server?
@contextmanager
def anki_connect_web_server_started():
plugin = AnkiConnect()
plugin.startWebServer()
yield plugin
@dataclass
class Client:
port: int
@staticmethod
def make_request(action, **params):
return {"action": action, "params": params, "version": 6}
def send_request(self, action, **params):
request_url = f"http://localhost:{self.port}"
request_data = self.make_request(action, **params)
request_json = json.dumps(request_data).encode("utf-8")
request = urllib.request.Request(request_url, request_json)
response = json.load(urllib.request.urlopen(request))
return response
def wait_for_web_server_to_come_live(self, at_most_seconds=30):
deadline = time.time() + at_most_seconds
while time.time() < deadline:
try:
self.send_request("version")
return
except urllib.error.URLError:
time.sleep(0.01)
raise Exception(f"Anki-Connect web server did not come live "
f"in {at_most_seconds} seconds")
# spawning requires a top-level function for pickling
def external_anki_entry_function(web_bind_port, exit_event):
with empty_anki_session_started() as session:
with anki_connect_config_loaded(session, web_bind_port):
with anki_connect_web_server_started():
with profile_loaded(session):
wait_until(exit_event.is_set)
@contextmanager
def external_anki_running(process_run_method):
context = multiprocessing.get_context(process_run_method)
exit_event = context.Event()
web_bind_port = find_free_port()
function = partial(external_anki_entry_function, web_bind_port, exit_event)
with function_running_in_a_process(context, function) as process:
client = Client(port=web_bind_port)
client.wait_for_web_server_to_come_live()
try:
yield client
finally:
exit_event.set()
assert process.exitcode == 0
# if a Qt app was already launched in current process,
# launching a new Qt app, even from grounds up, fails or hangs.
# of course, this includes forked processes. therefore,
# * if launching without --forked, use the `spawn` process run method;
# * otherwise, use the `fork` method, as it is significantly faster.
# with --forked, each test has its fixtures assembled inside the fork,
# which means that when the test begins, Qt was never started in the fork.
@pytest.fixture(scope="module")
def external_anki(request):
"""
Runs Anki in an external process, with the plugin loaded and started.
On exit, neatly ends the process and makes sure its exit code is 0.
Yields a client that can send web request to the external process.
"""
with external_anki_running(
"fork" if request.config.option.forked else "spawn"
) as client:
yield client
##############################################################################
def test_successful_request(external_anki):
response = external_anki.send_request("version")
assert response == {"error": None, "result": 6}
def test_can_handle_multiple_requests(external_anki):
assert external_anki.send_request("version") == {"error": None, "result": 6}
assert external_anki.send_request("version") == {"error": None, "result": 6}
def test_multi_request(external_anki):
version_request = Client.make_request("version")
response = external_anki.send_request("multi", actions=[version_request] * 3)
assert response == {
"error": None,
"result": [{"error": None, "result": 6}] * 3
}
def test_failing_request_due_to_bad_arguments(external_anki):
response = external_anki.send_request("addNote", bad="request")
assert response["result"] is None
assert "unexpected keyword argument" in response["error"]
def test_failing_request_due_to_anki_raising_exception(external_anki):
response = external_anki.send_request("suspend", cards=[-123])
assert response["result"] is None
assert "Card was not found" in response["error"]