Refactor network module - prepare it for easier sub-module testing

.github/workflows/integration-test.yml → .github/workflows/network-testing.yml

@@ -1,4 +1,4 @@
-name: Run ocrd network integration tests
+name: ocrd network module and integration tests
 
 on:
   push:
@@ -49,10 +49,12 @@ jobs:
         run: |
           brew install docker docker-compose
           colima start
-      - name: Test network integration with pytest
+      - name: Run network modules tests with pytest
+        run: make network-module-test
+      - name: Run network integration tests with pytest
         run: |
           if [[ "${{ matrix.os }}" == "macos"* ]];then
-            make integration-test DOCKER_COMPOSE=docker-compose
+            make network-integration-test-cicd DOCKER_COMPOSE=docker-compose
           else
-            make integration-test
+            make network-integration-test-cicd
           fi

.github/workflows/unit-test.yml

@@ -26,7 +26,7 @@ jobs:
         os:
           - ubuntu-22.04
           - ubuntu-20.04
-          - macos-latest
+          # - macos-latest
 
     steps:
     - uses: actions/checkout@v3

Makefile

@@ -39,6 +39,9 @@ help:
 	@echo "    docker         Build docker image"
 	@echo "    docker-cuda    Build docker image for GPU / CUDA"
 	@echo "    pypi           Build wheels and source dist and twine upload them"
+	@echo " ocrd network tests"
+	@echo "    network-module-test       Run all ocrd_network module tests"
+	@echo "    network-integration-test  Run all ocrd_network integration tests (docker and docker compose required)"
 	@echo ""
 	@echo "  Variables"
 	@echo ""
@@ -219,10 +222,21 @@ test: assets
 		$(TESTDIR)
 	cd ocrd_utils ; $(PYTHON) -m pytest --continue-on-collection-errors -k TestLogging -k TestDecorators $(TESTDIR)
 
+network-module-test: assets
+	$(PYTHON) \
+		-m pytest $(PYTEST_ARGS) -k 'test_modules_' -v --durations=10\
+		--ignore-glob="$(TESTDIR)/network/test_integration_*.py" \
+		$(TESTDIR)/network
+
 INTEGRATION_TEST_IN_DOCKER = docker exec core_test
-integration-test:
+network-integration-test:
+	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml up -d
+	-$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_integration_' -v
+	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml down --remove-orphans
+
+network-integration-test-cicd:
 	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml up -d
-	-$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_rmq or test_db or test_processing_server' -v
+	$(INTEGRATION_TEST_IN_DOCKER) pytest -k 'test_integration_' -v
 	$(DOCKER_COMPOSE) --file tests/network/docker-compose.yml down --remove-orphans
 
 benchmark:

README.md

@@ -80,7 +80,7 @@ A minimal [OCR-D processor](https://ocr-d.de/en/user_guide#using-the-ocr-d-proce
 
 Almost all behaviour of the OCR-D/core software is configured via CLI options and flags, which can be listed with the `--help` flag that all CLI support.
 
-Some parts of the software are configured via environement variables:
+Some parts of the software are configured via environment variables:
 
 * `OCRD_METS_CACHING`: If set to `true`, access to the METS file is cached, speeding in-memory search and modification.
 * `OCRD_PROFILE`: This variable configures the built-in CPU and memory profiling. If empty, no profiling is done. Otherwise expected to contain any of the following tokens:
@@ -105,7 +105,7 @@ Some parts of the software are configured via environement variables:
 * `OCRD_NETWORK_SERVER_ADDR_PROCESSING`: Default address of Processing Server to connect to (for `ocrd network client processing`).
 * `OCRD_NETWORK_SERVER_ADDR_WORKFLOW`: Default address of Workflow Server to connect to (for `ocrd network client workflow`).
 * `OCRD_NETWORK_SERVER_ADDR_WORKSPACE`: Default address of Workspace Server to connect to (for `ocrd network client workspace`).
-* `OCRD_NETWORK_WORKER_QUEUE_CONNECT_ATTEMPTS`: Number of attempts for a worker to create its queue. Helpfull if the rabbitmq-server needs time to be fully started.
+* `OCRD_NETWORK_RABBITMQ_CLIENT_CONNECT_ATTEMPTS`: Number of attempts for a worker to create its queue. Helpfull if the rabbitmq-server needs time to be fully started.
 
 
 ## Packages

src/ocrd/cli/__init__.py

@@ -43,7 +43,7 @@
 \b
 {config.describe('OCRD_NETWORK_SERVER_ADDR_WORKSPACE')}
 \b
-{config.describe('OCRD_NETWORK_WORKER_QUEUE_CONNECT_ATTEMPTS')}
+{config.describe('OCRD_NETWORK_RABBITMQ_CLIENT_CONNECT_ATTEMPTS')}
 \b
 {config.describe('OCRD_PROFILE_FILE')}
 \b

src/ocrd/cli/network.py

@@ -7,7 +7,6 @@
 """
 
 import click
-import logging
 from ocrd_utils import initLogging
 from ocrd_network.cli import (
     client_cli,

src/ocrd/decorators/__init__.py

@@ -10,7 +10,7 @@
     set_json_key_value_overrides,
 )
 from ocrd_validators import WorkspaceValidator
-from ocrd_network import ProcessingWorker, ProcessorServer, NETWORK_AGENT_SERVER, NETWORK_AGENT_WORKER
+from ocrd_network import ProcessingWorker, ProcessorServer, AgentType
 
 from ..resolver import Resolver
 from ..processor.base import run_processor
@@ -20,7 +20,7 @@
 from .ocrd_cli_options import ocrd_cli_options
 from .mets_find_options import mets_find_options
 
-SUBCOMMANDS = [NETWORK_AGENT_WORKER, NETWORK_AGENT_SERVER]
+SUBCOMMANDS = [AgentType.PROCESSING_WORKER, AgentType.PROCESSOR_SERVER]
 
 
 def ocrd_cli_wrap_processor(
@@ -142,19 +142,19 @@ def check_and_run_network_agent(ProcessorClass, subcommand: str, address: str, d
     if not database:
         raise ValueError(f"Option '--database' is invalid for subcommand {subcommand}")
 
-    if subcommand == NETWORK_AGENT_SERVER:
+    if subcommand == AgentType.PROCESSOR_SERVER:
         if not address:
             raise ValueError(f"Option '--address' required for subcommand {subcommand}")
         if queue:
             raise ValueError(f"Option '--queue' invalid for subcommand {subcommand}")
-    if subcommand == NETWORK_AGENT_WORKER:
+    if subcommand == AgentType.PROCESSING_WORKER:
         if address:
             raise ValueError(f"Option '--address' invalid for subcommand {subcommand}")
         if not queue:
             raise ValueError(f"Option '--queue' required for subcommand {subcommand}")
 
     processor = ProcessorClass(workspace=None)
-    if subcommand == NETWORK_AGENT_WORKER:
+    if subcommand == AgentType.PROCESSING_WORKER:
         processing_worker = ProcessingWorker(
             rabbitmq_addr=queue,
             mongodb_addr=database,
@@ -166,7 +166,7 @@ def check_and_run_network_agent(ProcessorClass, subcommand: str, address: str, d
         processing_worker.connect_consumer()
         # Start consuming from the queue with name `processor_name`
         processing_worker.start_consuming()
-    elif subcommand == NETWORK_AGENT_SERVER:
+    elif subcommand == AgentType.PROCESSOR_SERVER:
         # TODO: Better validate that inside the ProcessorServer itself
         host, port = address.split(':')
         processor_server = ProcessorServer(
@@ -175,4 +175,6 @@ def check_and_run_network_agent(ProcessorClass, subcommand: str, address: str, d
             processor_class=ProcessorClass,
         )
         processor_server.run_server(host=host, port=int(port))
+    else:
+        raise ValueError(f"Unknown network agent type, must be one of: {SUBCOMMANDS}")
     sys.exit(0)

src/ocrd/decorators/ocrd_cli_options.py

@@ -1,7 +1,7 @@
 import click
 from click import option, Path, group, command, argument
 from ocrd_utils import DEFAULT_METS_BASENAME
-from ocrd_network import NETWORK_AGENT_SERVER, NETWORK_AGENT_WORKER
+from ocrd_network import AgentType
 from .parameter_option import parameter_option, parameter_override_option
 from .loglevel_option import loglevel_option
 from ocrd_network import (
@@ -57,7 +57,7 @@ def cli(mets_url):
         # subcommands. So we have to work around that by creating a
         # pseudo-subcommand handled in ocrd_cli_wrap_processor
         argument('subcommand', nargs=1, required=False,
-                 type=click.Choice([NETWORK_AGENT_WORKER, NETWORK_AGENT_SERVER])),
+                 type=click.Choice([AgentType.PROCESSING_WORKER, AgentType.PROCESSOR_SERVER])),
     ]
     for param in params:
         param(f)

src/ocrd_network/__init__.py

@@ -1,10 +1,7 @@
 from .client import Client
-from .constants import NETWORK_AGENT_SERVER, NETWORK_AGENT_WORKER
+from .constants import AgentType, JobState
 from .processing_server import ProcessingServer
 from .processing_worker import ProcessingWorker
 from .processor_server import ProcessorServer
-from .param_validators import (
-    DatabaseParamType,
-    ServerAddressParamType,
-    QueueServerParamType
-)
+from .param_validators import DatabaseParamType, ServerAddressParamType, QueueServerParamType
+from .server_cache import CacheLockedPages, CacheProcessingRequests

src/ocrd_network/cli/client.py

@@ -1,13 +1,9 @@
 import click
 from typing import Optional
 
-from ocrd_utils import DEFAULT_METS_BASENAME
-
-from ocrd.decorators import (
-    parameter_option,
-    parameter_override_option
-)
+from ocrd.decorators import parameter_option
 from ocrd_network import Client
+from ocrd_utils import DEFAULT_METS_BASENAME
 
 
 @click.group('client')

src/ocrd_network/cli/processing_server.py

@@ -1,8 +1,5 @@
 import click
-from .. import (
-    ProcessingServer,
-    ServerAddressParamType
-)
+from ocrd_network import ProcessingServer, ServerAddressParamType
 
 
 @click.command('processing-server')

src/ocrd_network/cli/processing_worker.py

@@ -1,11 +1,6 @@
 import click
 from ocrd_utils import get_ocrd_tool_json
-
-from .. import (
-    DatabaseParamType,
-    ProcessingWorker,
-    QueueServerParamType
-)
+from ocrd_network import DatabaseParamType, ProcessingWorker, QueueServerParamType
 
 
 @click.command('processing-worker')

src/ocrd_network/cli/processor_server.py

@@ -1,9 +1,5 @@
 import click
-from .. import (
-    DatabaseParamType,
-    ProcessorServer,
-    ServerAddressParamType
-)
+from ocrd_network import DatabaseParamType, ProcessorServer, ServerAddressParamType
 
 
 @click.command('processor-server')
@@ -23,12 +19,12 @@ def processor_server_cli(processor_name: str, address: str, database: str):
     (standalone REST API OCR-D processor)
     """
     try:
-        # TODO: Better validate that inside the ProcessorServer itself
+        # Note, the address is already validated with the type field
         host, port = address.split(':')
         processor_server = ProcessorServer(
             mongodb_addr=database,
             processor_name=processor_name,
-            processor_class=None,  # For readability purposes assigned here
+            processor_class=None  # For readability purposes assigned here
         )
         processor_server.run_server(host=host, port=int(port))
     except Exception as e:

src/ocrd_network/client.py

@@ -1,38 +1,37 @@
-import json
-import requests
+from json import dumps, loads
+from requests import post as requests_post
+from ocrd_utils import config, getLogger, LOG_FORMAT
 
-from ocrd_utils import config
+from .constants import NETWORK_PROTOCOLS
 
 
 # TODO: This is just a conceptual implementation and first try to
 #  trigger further discussions on how this should look like.
 class Client:
     def __init__(
-            self,
-            server_addr_processing: str = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING,
-            server_addr_workflow: str = config.OCRD_NETWORK_SERVER_ADDR_WORKFLOW,
-            server_addr_workspace: str = config.OCRD_NETWORK_SERVER_ADDR_WORKSPACE,
+        self,
+        server_addr_processing: str = config.OCRD_NETWORK_SERVER_ADDR_PROCESSING,
+        server_addr_workflow: str = config.OCRD_NETWORK_SERVER_ADDR_WORKFLOW,
+        server_addr_workspace: str = config.OCRD_NETWORK_SERVER_ADDR_WORKSPACE
     ):
+        self.log = getLogger(f"ocrd_network.client")
         self.server_addr_processing = server_addr_processing
         self.server_addr_workflow = server_addr_workflow
         self.server_addr_workspace = server_addr_workspace
 
     def send_processing_request(self, processor_name: str, req_params: dict):
         verify_server_protocol(self.server_addr_processing)
-        req_url = f'{self.server_addr_processing}/processor/{processor_name}'
+        req_url = f"{self.server_addr_processing}/processor/{processor_name}"
         req_headers = {"Content-Type": "application/json; charset=utf-8"}
-        req_json = json.loads(json.dumps(req_params))
-
-        print(f'Sending processing request to: {req_url}')
-        response = requests.post(url=req_url, headers=req_headers, json=req_json)
+        req_json = loads(dumps(req_params))
+        self.log.info(f"Sending processing request to: {req_url}")
+        self.log.debug(req_json)
+        response = requests_post(url=req_url, headers=req_headers, json=req_json)
         return response.json()
 
 
 def verify_server_protocol(address: str):
-    protocol_matched = False
-    for protocol in ['http://', 'https://']:
+    for protocol in NETWORK_PROTOCOLS:
         if address.startswith(protocol):
-            protocol_matched = True
-            break
-    if not protocol_matched:
-        raise ValueError(f'Wrong/Missing protocol in the server address: {address}')
+            return
+    raise ValueError(f"Wrong/Missing protocol in the server address: {address}, must be one of: {NETWORK_PROTOCOLS}")

src/ocrd_network/constants.py

@@ -1,2 +1,59 @@
-NETWORK_AGENT_SERVER = 'server'
-NETWORK_AGENT_WORKER = 'worker'
+from enum import Enum
+
+DOCKER_IMAGE_MONGO_DB = "mongo"
+DOCKER_IMAGE_RABBIT_MQ = "rabbitmq:3.12-management"
+# These feature flags are required by default to use the newer version
+DOCKER_RABBIT_MQ_FEATURES = "quorum_queue,implicit_default_bindings,classic_mirrored_queue_version"
+
+NETWORK_PROTOCOLS = ["http://", "https://"]
+OCRD_ALL_JSON_TOOLS_URL = "https://ocr-d.de/js/ocrd-all-tool.json"
+SERVER_ALL_PAGES_PLACEHOLDER = "all_pages"
+
+
+class AgentType(str, Enum):
+    PROCESSING_WORKER = "worker"
+    PROCESSOR_SERVER = "server"
+
+
+class DeployType(str, Enum):
+    # Deployed by the Processing Server config file
+    DOCKER = "docker"
+    NATIVE = "native"
+    # Deployed through a registration endpoint of the Processing Server
+    # TODO: That endpoint is still not implemented
+    EXTERNAL = "external"
+
+
+# TODO: Make the states uppercase
+class JobState(str, Enum):
+    # The processing job is cached inside the Processing Server requests cache
+    cached = "CACHED"
+    # The processing job was cancelled due to failed dependencies
+    cancelled = "CANCELLED"
+    # Processing job failed
+    failed = "FAILED"
+    # The processing job is queued inside the RabbitMQ
+    queued = "QUEUED"
+    # Processing job is currently running in a Worker or Processor Server
+    running = "RUNNING"
+    # Processing job finished successfully
+    success = "SUCCESS"
+    # Processing job has not been assigned yet
+    unset = "UNSET"
+
+
+class NetworkLoggingDirs(str, Enum):
+    METS_SERVERS = "mets_servers"
+    PROCESSING_JOBS = "processing_jobs"
+    PROCESSING_SERVERS = "processing_servers"
+    PROCESSING_WORKERS = "processing_workers"
+    PROCESSOR_SERVERS = "processor_servers"
+
+
+class ServerApiTags(str, Enum):
+    ADMIN = "admin"
+    DISCOVERY = "discovery"
+    PROCESSING = "processing"
+    TOOLS = "tools"
+    WORKFLOW = "workflow"
+    WORKSPACE = "workspace"