diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml index 4c71b48d3447ba4bf3186928b5a6fdc6b96a4c42..e70771c2cd1ba710785f6968318360cdad5fbf98 100644 --- a/.gitlab-ci.yml +++ b/.gitlab-ci.yml @@ -1,90 +1,65 @@ variables: WORDPRESS_VERSION: - value: 5.8.1 + value: 6.0.3 description: WordPress release PHP_VERSION: - value: 8.0.0 + value: 8.1.11 description: PHP release to build into the backend image NGINX_VERSION: - value: 1.21.3 + value: 1.23.2 description: Nginx release for the frontend image -.changes: &change-files - changes: - - .gitlab-ci.yml - - Dockerfile - - data/* - - plugins/* - - scripts/* - -.build: - stage: build - image: docker.kodo.org.uk/ci-images/buildkit/buildctl:latest - tags: [buildkit] +workflow: rules: - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH + variables: {BUILD_RELEASE: "true"} - if: $CI_PIPELINE_SOURCE == "schedule" - - if: $CI_PIPELINE_SOURCE == "merge_request_event" - - << : *change-files - script: - - BUILD_TAG=${CI_REGISTRY_IMAGE}/${TARGET}:build-${CI_PIPELINE_IID} - - buildctl build - --frontend=dockerfile.v0 - --local context=. - --local dockerfile=. - --opt target=${TARGET} - --opt build-arg:nginx_version=${NGINX_VERSION} - --opt build-arg:php_version=${PHP_VERSION} - --opt build-arg:wp_version=${WORDPRESS_VERSION} - --opt label:nginx.version=${NGINX_VERSION} - --opt label:php.version=${PHP_VERSION} - --opt label:wordpress.version=${WORDPRESS_VERSION} - --output type=image,name=${BUILD_TAG},push=true + variables: {BUILD_RELEASE: "true"} + - when: always -.tag: - stage: deploy - image: docker.kodo.org.uk/ci-images/docker-reg:latest - rules: - - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH && $CI_PIPELINE_SOURCE == "schedule" - variables: - TAG: latest - - if: $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH - << : *change-files - variables: - TAG: latest - - if: $CI_COMMIT_BRANCH == "develop" - << : *change-files - variables: - TAG: unstable - script: | - BUILD_TAG=${CI_REGISTRY_IMAGE}/${TARGET}:build-${CI_PIPELINE_IID} - docker-reg $BUILD_TAG retag $TAG - if [ $CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH ]; then - docker-reg $BUILD_TAG retag $VERSION - fi +include: +- project: dom/project-templates + file: /pipeline-templates/pre-commit.yml -Build Wordpress: - extends: [.build] - variables: - TARGET: fastcgi - -Build Nginx: - extends: [.build] +Build Images: + stage: build + rules: + - if: $BUILD_RELEASE + when: never + - if: $CI_PIPELINE_SOURCE == "web" + - changes: + - .gitlab-ci.yml + - Dockerfile + - data/* + - plugins/* + - scripts/* variables: - TARGET: nginx - + PLATFORMS: "" + BUILD_ARGS: >- + wp_version=$WORDPRESS_VERSION + php_version=$PHP_VERSION + nginx_version=$NGINX_VERSION + parallel: + matrix: + - TARGET: [nginx, fastcgi] + trigger: + include: + - project: dom/project-templates + file: /pipeline-templates/build-image.yml + strategy: depend -Tag Wordpress Image: - extends: [.tag] - variables: - TARGET: fastcgi - VERSION: $WORDPRESS_VERSION -Tag Nginx Image: - extends: [.tag] - variables: - TARGET: nginx - VERSION: $NGINX_VERSION +Build Releases: + stage: build + extends: [Build Images] + rules: + - if: $BUILD_RELEASE + parallel: + matrix: + - TARGET: [nginx] + RELEASE: [$NGINX_VERSION] + - TARGET: [fastcgi] + RELEASE: [$WORDPRESS_VERSION] diff --git a/.lint.cfg b/.lint.cfg index c56b8d61f60b8ee811c8b58ffeed7bd3d4de702e..ae899631ae3ee2fd3cbbbfe18995be86acda4da2 100644 --- a/.lint.cfg +++ b/.lint.cfg @@ -5,9 +5,7 @@ force_single_line = true strict = true warn_unused_configs = true warn_unreachable = true -mypy_path = tests, tests/stubs -plugins = - trio_typing.plugin +mypy_path = tests [flake8] max-line-length = 92 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index e7292c36580f9c2ad6e27cc686095f670e95711c..f526eaa9df978dac0b4e63c253f77c2115d6af9e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -40,7 +40,7 @@ repos: - id: check-executable-modes - id: check-for-squash - id: copyright-notice - exclude: ^data/ + exclude: ^data/|^scripts/(compile-|install-) - id: protect-first-parent - repo: https://github.com/pre-commit/pygrep-hooks @@ -95,6 +95,5 @@ repos: - id: mypy args: ["--config-file=.lint.cfg"] additional_dependencies: - - trio-typing - types-requests - - git+https://code.kodo.org.uk/dom/type-stubs.git#type-stubs[jsonpath,parse] + - behave-utils ~=0.3.2 diff --git a/data/composer.json b/data/composer.json index 51f5bb2ee44a45dde0c1bcc81c94eb14dff1d9cb..a2ed22547472c5a6eddcd9eb2305c8cfbff4ad75 100644 --- a/data/composer.json +++ b/data/composer.json @@ -12,6 +12,9 @@ "ayesh/wordpress-password-hash": "2.*" }, "config": { + "allow-plugins": { + "composer/installers": true + }, "gitlab-domains": [ "code.kodo.org.uk" ] }, "extra": { diff --git a/data/nginx/502.html b/data/nginx/html/502.html similarity index 100% rename from data/nginx/502.html rename to data/nginx/html/502.html diff --git a/data/nginx/server.conf b/data/nginx/server.conf index afc77233eb1802a70c98f7c4d092acb30b4a6cda..48a685a05ab88c436c46d1863d26086e9aebefba 100644 --- a/data/nginx/server.conf +++ b/data/nginx/server.conf @@ -49,11 +49,6 @@ server { access_log off; } - # Don't return 200 for a missing favicon - location = /favicon.ico { - try_files favicon.ico =404; - } - # Don't delegate to index.php for /.well-known/ # If a plugin wants to handle /.well-known/ URIs please submit an issue to # https://code.kodo.org.uk/singing-chimes.co.uk/wordpress/ @@ -95,6 +90,14 @@ server { location /wp-admin/ { try_files $uri $uri/index.php; + location = /wp-admin/load-styles.php { + return 404; + } + + location = /wp-admin/load-scripts.php { + return 404; + } + location ~ \.php$ { include fastcgi-script.conf; include cache-bust.conf; diff --git a/data/wp-config.php b/data/wp-config.php index 4a6a728e70d5a8163aefa9d55ace8505529e840d..ce4b0c4a71b93580946dc431b3d1c7d96604f069 100644 --- a/data/wp-config.php +++ b/data/wp-config.php @@ -16,6 +16,13 @@ define('DISABLE_WP_CRON', true); **/ define('UPLOADS', 'media'); +/** + * Disable script concatenation in the admin interface. + * This puts extra load on the PHP server that Nginx should be taking. + * The benefits of concatenation should be negated when using HTTP/2. + **/ +define('CONCATENATE_SCRIPTS', false); + /** * Stop the site-health tool from complaining about unwritable filesystems. * Background upgrades are performed by a user with write privileges via the diff --git a/doc/configuration.md b/doc/configuration.md index 95e988749c10d60bef6cad95421fd5350acbdff4..bfa468d9c27bb84ffff881de68bc6e7ddd00f5d1 100644 --- a/doc/configuration.md +++ b/doc/configuration.md @@ -237,7 +237,7 @@ The path to a file containing lines to append to [**THEMES**](#themes). **Type**: array\ **Required**: no\ -**Default**: /etc/wordpress/*config.php +**Default**: /etc/wordpress/**/*config.php This is an array of files to include in wp-config.php. The default includes a wildcard which is expanded. Wildcards in the environment variable will diff --git a/scripts/compile-imagick.sh b/scripts/compile-imagick.sh index ab505410cefe85370b8903eb44715b28ba74a496..f75695ab18412f57633581f3f54ad71a836d0c68 100755 --- a/scripts/compile-imagick.sh +++ b/scripts/compile-imagick.sh @@ -1,10 +1,4 @@ #!/bin/bash -# Copyright 2019-2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - set -eux shopt -s lastpipe diff --git a/scripts/entrypoint.sh b/scripts/entrypoint.sh index b552687f819da5b83fd6dbdfbf664d09145e7464..d8c813faf4d7f6e4b816bdef43eaf290ef5702fb 100755 --- a/scripts/entrypoint.sh +++ b/scripts/entrypoint.sh @@ -1,6 +1,6 @@ #!/bin/bash # -# Copyright 2019-2021 Dominik Sekotill +# Copyright 2019-2022 Dominik Sekotill # # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this @@ -13,7 +13,7 @@ shopt -s nullglob globstar extglob enable -f /usr/lib/bash/head head enable -f /usr/lib/bash/unlink unlink -declare -r DEFAULT_THEME=twentytwentyone +declare -r DEFAULT_THEME=twentytwentytwo declare -r WORKER_USER=www-data declare -r CONFIG_DIR=/etc/wordpress declare -r WORK_DIR=${PWD} @@ -42,8 +42,7 @@ declare -a PHP_DIRECTIVES=( post_max_size=20M ) declare -a WP_CONFIGS=( - ${WP_CONFIGS-} - ${CONFIG_DIR}/*config.php + ${WP_CONFIGS-${CONFIG_DIR}/**/*config.php} ) diff --git a/scripts/install-build-deps.sh b/scripts/install-build-deps.sh index 5b7e7e1bad58f08c6e8e603b2e2025f48518fc57..bee79d377dfbe030fdafe20ff95d28b2e05bd230 100755 --- a/scripts/install-build-deps.sh +++ b/scripts/install-build-deps.sh @@ -1,10 +1,4 @@ #!/bin/sh -# Copyright 2019-2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - set -eux # Install packaged dependencies diff --git a/scripts/install-deps.sh b/scripts/install-deps.sh index 5dd7e75fb774ece0370e004f64c91c0a53abb120..3456cab272006ffd9b4de03e00d8fd87f8d87a4c 100755 --- a/scripts/install-deps.sh +++ b/scripts/install-deps.sh @@ -1,10 +1,4 @@ #!/bin/sh -# Copyright 2019-2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - set -eux # Install packaged dependencies diff --git a/scripts/install-wp.sh b/scripts/install-wp.sh index 32f27dfd5a79779037ee4da76c8f4cf77cd7e58d..c2a7f27c1e4b8c5db3e2746a6649094c2d1a9108 100755 --- a/scripts/install-wp.sh +++ b/scripts/install-wp.sh @@ -1,10 +1,4 @@ #!/bin/bash -# Copyright 2019-2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - set -eux COMPOSER_INSTALLER_URL=https://getcomposer.org/installer diff --git a/tests/environment.py b/tests/environment.py index ee5727fe1121d5f3558e437608675d371a65e7c6..1d0837e813d12b580f5060234ec56857746f2ed2 100644 --- a/tests/environment.py +++ b/tests/environment.py @@ -15,36 +15,28 @@ https://behave.readthedocs.io/en/stable/tutorial.html#environmental-controls from __future__ import annotations import sys -from contextlib import contextmanager from os import environ -from pathlib import Path from typing import TYPE_CHECKING from typing import Any from typing import Iterator -from typing import NamedTuple from behave import fixture from behave import use_fixture from behave.model import Feature from behave.model import Scenario from behave.runner import Context +from behave_utils import URL +from behave_utils import redirect +from behave_utils.mysql import snapshot_rollback from requests.sessions import Session -from utils import URL -from utils import make_secret -from utils import redirect -from wp import Mysql -from wp import Wordpress -from wp.docker import Container -from wp.docker import Image -from wp.docker import IPv4Address -from wp.docker import Network +from wp import Site +from wp import test_cluster if TYPE_CHECKING: from behave.runner import FeatureContext from behave.runner import ScenarioContext SITE_URL = URL("http://test.example.com") -BUILD_CONTEXT = Path(__file__).parent.parent def before_all(context: Context) -> None: @@ -58,7 +50,7 @@ def before_feature(context: FeatureContext, feature: Feature) -> None: """ Setup/revert fixtures before each feature """ - use_fixture(db_snapshot_rollback, context) + use_fixture(snapshot_rollback, context, context.site.database) def before_scenario(context: ScenarioContext, scenario: Scenario) -> None: @@ -68,23 +60,11 @@ def before_scenario(context: ScenarioContext, scenario: Scenario) -> None: context.session = use_fixture(requests_session, context) -class Site(NamedTuple): - """ - A named-tuple of information about the containers for a site fixture - """ - - url: str - address: IPv4Address - frontend: Container - backend: Wordpress - database: Mysql - - # Todo(dom.sekotill): When PEP-612 is properly implemented in mypy the [*a, **k] and default # values nonsense can be removed from fixtures @fixture -def setup_test_cluster(context: Context, /, site_url: str|None = None, *a: Any, **k: Any) -> Iterator[Site]: +def setup_test_cluster(context: Context, /, site_url: URL|None = None, *a: Any, **k: Any) -> Iterator[Site]: """ Prepare and return the details of a site fixture """ @@ -116,71 +96,6 @@ def db_snapshot_rollback(context: FeatureContext, /, *a: Any, **k: Any) -> Itera db.mysql(input=snapshot) -@contextmanager -def test_cluster(site_url: str) -> Iterator[Site]: - """ - Configure and start all the necessary containers for use as test fixtures - """ - test_dir = Path(__file__).parent - - db_secret = make_secret(20) - db_init = test_dir / "mysql-init.sql" - - with Network() as network: - database = Mysql( - Image.pull("mysql/mysql-server"), - network=network, - volumes={ - Path("/var/lib/mysql"), - (db_init, Path("/docker-entrypoint-initdb.d") / db_init.name), - }, - env=dict( - MYSQL_DATABASE="test-db", - MYSQL_USER="test-db-user", - MYSQL_PASSWORD=db_secret, - ), - ) - frontend = Container( - Image.build( - BUILD_CONTEXT, - target='nginx', - nginx_version=environ.get("NGINX_VERSION"), - ), - network=network, - volumes=[ - ("static", Path("/app/static")), - ("media", Path("/app/media")), - ], - ) - backend = Wordpress( - Image.build( - BUILD_CONTEXT, - php_version=environ.get("PHP_VERSION"), - wp_version=environ.get("WP_VERSION"), - ), - network=network, - volumes=frontend.volumes, - env=dict( - SITE_URL=site_url, - SITE_ADMIN_EMAIL="test@kodo.org.uk", - DB_NAME="test-db", - DB_USER="test-db-user", - DB_PASS=db_secret, - DB_HOST="database:3306", - ), - ) - - backend.connect(network, "upstream") - database.connect(network, "database") - - with database.started(), backend.started(), frontend.started(): - addr = frontend.inspect( - f"$.NetworkSettings.Networks.{network}.IPAddress", - str, IPv4Address, - ) - yield Site(site_url, addr, frontend, backend, database) - - if __name__ == "__main__": from subprocess import run diff --git a/tests/regression-20.feature b/tests/regression-20.feature new file mode 100644 index 0000000000000000000000000000000000000000..56a020c61b55fcd69d2b7ebf1c7524cb4fa0f660 --- /dev/null +++ b/tests/regression-20.feature @@ -0,0 +1,15 @@ +Feature: Return favicons when requests + Regression check for "#20": Fix/remove favicon.ico intercept in Nginx + + An icon should always be served when requesting "/favicon.ico", either one + added by the owner, or a default icon. + + Scenario: Default icon + When /favicon.ico is requested + Then 302 is returned + And the "Location" header's value is "http://test.example.com/wp-includes/images/w-logo-blue-white-bg.png" + + Scenario: Owner supplied icon + Given /app/static/favicon.ico exists in the frontend + When /favicon.ico is requested + Then 200 is returned diff --git a/tests/requirements.txt b/tests/requirements.txt index 77b00734a9b6bf9c5eb9e68a03386ee4569b7352..bb41f9e7b4c6a446ef3f5b5105e5181b28bea9ac 100644 --- a/tests/requirements.txt +++ b/tests/requirements.txt @@ -1,6 +1,5 @@ Python ~=3.9; python_version < '3.9' behave -jsonpath-python ~=1.0 +behave-utils ~=0.3.2 requests ~=2.26 -trio ~=0.19.0 diff --git a/tests/steps/commands.py b/tests/steps/commands.py index 897f21e9f6de4657aa95538ee7db6302f1eca920..422a8f7a6bb0343d4666afea6a9abe9020dd1fb0 100644 --- a/tests/steps/commands.py +++ b/tests/steps/commands.py @@ -16,8 +16,8 @@ from typing import TYPE_CHECKING from behave import then from behave import when -from utils.behave import PatternEnum -from utils.behave import register_pattern +from behave_utils.behave import PatternEnum +from behave_utils.behave import register_pattern from wp import Container if TYPE_CHECKING: diff --git a/tests/steps/pages.py b/tests/steps/pages.py index 151e78cbf254868f3e8451f4a2359cc9bab7bca0..7921b4f2a00d1c0135d79d3a4f5e93b6e81a1f22 100644 --- a/tests/steps/pages.py +++ b/tests/steps/pages.py @@ -20,11 +20,11 @@ from behave import then from behave import use_fixture from behave import when from behave.runner import Context +from behave_utils import URL +from behave_utils import JSONArray +from behave_utils import JSONObject +from behave_utils import PatternEnum from request_steps import get_request -from utils import URL -from utils import JSONArray -from utils import JSONObject -from utils import PatternEnum DEFAULT_CONTENT = """ Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut diff --git a/tests/steps/request_steps.py b/tests/steps/request_steps.py index 6af8d985e50370ff340cc3bbd092a61777709be5..211c4d4f20c4a7bfd79d4ac85f176c9d437d8f0e 100644 --- a/tests/steps/request_steps.py +++ b/tests/steps/request_steps.py @@ -1,4 +1,4 @@ -# Copyright 2021 Dominik Sekotill +# Copyright 2021-2022 Dominik Sekotill # # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this @@ -16,8 +16,10 @@ from typing import Any from behave import then from behave import when from behave.runner import Context -from utils import URL -from utils import PatternEnum +from behave_utils import URL +from behave_utils import PatternEnum + +SAMPLE_SITE_NAME = "http://test.example.com" class Method(PatternEnum): @@ -120,6 +122,8 @@ def assert_header(context: Context, header_name: str, header_value: str) -> None """ Assert that an expected header was received during a previous step """ + if SAMPLE_SITE_NAME in header_value: + header_value = header_value.replace(SAMPLE_SITE_NAME, context.site.url) headers = context.response.headers assert header_name in headers, \ f"Expected header not found in response: {header_name!r}" diff --git a/tests/steps/static.py b/tests/steps/static.py new file mode 100644 index 0000000000000000000000000000000000000000..b64dbe85e8af0bc54f8bb3a2ddac347c9601a57e --- /dev/null +++ b/tests/steps/static.py @@ -0,0 +1,39 @@ +# Copyright 2022 Dominik Sekotill +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +""" +Step implementations involving creating files in containers +""" + +from __future__ import annotations + +from typing import Iterator + +from behave import fixture +from behave import given +from behave import use_fixture +from behave.runner import Context +from behave_utils.docker import Cli + + +@given("{path} exists in the {container}") +def step_impl(context: Context, path: str, container: str) -> None: + """ + Create a file in the named container + """ + use_fixture(container_file, context, path, container) + + +@fixture +def container_file(context: Context, path: str, container_name: str) -> Iterator[None]: + """ + Create a file in a named container as a fixture + """ + container = getattr(context.site, container_name) + run = Cli(container) + run("tee", path, input=getattr(context, "text", "This is a data file!")) + yield + run("rm", path) diff --git a/tests/stubs/behave/__init__.pyi b/tests/stubs/behave/__init__.pyi deleted file mode 100644 index 289273d53bdb0b40e7b62731c4e267acd99c0ce9..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/__init__.pyi +++ /dev/null @@ -1,30 +0,0 @@ -from .fixture import fixture -from .fixture import use_fixture -from .matchers import register_type -from .matchers import use_step_matcher -from .step_registry import Given -from .step_registry import Step -from .step_registry import Then -from .step_registry import When -from .step_registry import given -from .step_registry import step -from .step_registry import then -from .step_registry import when - -# .matchers.step_matcher is deprecated, not including - -__version__: str -__all__ = [ - "Given", - "Step", - "Then", - "When", - "fixture", - "given", - "register_type", - "step", - "then", - "use_fixture", - "use_step_matcher", - "when", -] diff --git a/tests/stubs/behave/capture.pyi b/tests/stubs/behave/capture.pyi deleted file mode 100644 index 8bf42bfc66c6ec3992d4c50261897b8c43e40a27..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/capture.pyi +++ /dev/null @@ -1,43 +0,0 @@ -from io import StringIO -from typing import IO - -from .log_capture import LoggingCapture -from .model import Configuration -from .runner import Context - -class Captured: - - stdout: str - stderr: str - log_output: str - - @property - def output(self) -> str: ... - - def __init__(self, stdout: str = ..., stderr: str = ..., log_output: str = ...): ... - def __add__(self, other: Captured) -> Captured: ... - - def reset(self) -> None: ... - def add(self, captured: Captured) -> Captured: ... - def make_report(self) -> str: ... - - -class CaptureController: - - config: Configuration - stdout_capture: StringIO - stderr_capture: StringIO - log_capture: LoggingCapture - old_stdout: IO[str] - old_stderr: IO[str] - - @property - def capture(self) -> Captured: ... - - def __init__(self, config: Configuration): ... - - def setup_capture(self, context: Context) -> None: ... - def start_capture(self) -> None: ... - def stop_capture(self) -> None: ... - def teardown_capture(self) -> None: ... - def make_capture_report(self) -> str: ... diff --git a/tests/stubs/behave/fixture.pyi b/tests/stubs/behave/fixture.pyi deleted file mode 100644 index 4af0fab1073053d19ca42c53ba699b253b513515..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/fixture.pyi +++ /dev/null @@ -1,118 +0,0 @@ -import sys -from typing import Any -from typing import Iterator -from typing import Protocol -from typing import TypeVar -from typing import overload - -from .runner import Context - -C = TypeVar("C", bound=Context) -C_con = TypeVar("C_con", bound=Context, contravariant=True) - -R = TypeVar("R") -R_co = TypeVar("R_co", covariant=True) - - -# There's a lot of @overload-ed functions here as fixtures come in two varieties: -# 1) A @contextlib.contextmanager-like generator that yields an arbitrary object once. -# 2) A simple function that returns an arbitrary object -# -# "use_fixture" allows both types of fixture callables to be used in the same way - -if sys.version_info >= (3, 10) and False: - # This depends on complete support of ParamSpec in mypy so is disabled for now. - - from typing import ParamSpec - - P = ParamSpec("P") - - - class FixtureCoroutine(Protocol[C_con, P, R_co]): - def __call__(self, _: C_con, /, *__a: P.args, **__k: P.kwargs) -> Iterator[R_co]: ... - - class FixtureFunction(Protocol[C_con, P, R_co]): - def __call__(self, _: C_con, /, *__a: P.args, **__k: P.kwargs) -> R_co: ... - - - @overload - def use_fixture( - fixture_func: FixtureCoroutine[C_con, P, R], - context: C_con, - *a: P.args, - **k: P.kwargs, - ) -> R: ... - - @overload - def use_fixture( - fixture_func: FixtureFunction[C_con, P, R], - context: C_con, - *a: P.args, - **k: P.kwargs, - ) -> R: ... - -else: - # Without ParamSpec no checking is done to ensure the arguments passed to use_fixture - # match the fixture's arguments; fixtures must be able to handle arguments not being - # supplied (except the context); and fixtures must accept ANY arbitrary keyword - # arguments. - - P = TypeVar("P", bound=None) - P_co = TypeVar("P_co", covariant=True) # unused - - - class FixtureCoroutine(Protocol[C_con, P_co, R_co]): - def __call__(self, _: C_con, /, *__a: Any, **__k: Any) -> Iterator[R_co]: ... - - class FixtureFunction(Protocol[C_con, P_co, R_co]): - def __call__(self, _: C_con, /, *__a: Any, **__k: Any) -> R_co: ... - - - @overload - def use_fixture( - fixture_func: FixtureCoroutine[C_con, P_co, R_co], - context: C_con, - *a: Any, - **k: Any, - ) -> R_co: ... - - @overload - def use_fixture( - fixture_func: FixtureFunction[C_con, P_co, R_co], - context: C_con, - *a: Any, - **k: Any, - ) -> R_co: ... - - -# "fixture" is a decorator used to mark both types of fixture callables. It can also return -# a decorator, when called without the "func" argument. - -@overload -def fixture( - func: FixtureCoroutine[C, P, R], - name: str = ..., - pattern: str = ..., -) -> FixtureCoroutine[C, P, R]: ... - -@overload -def fixture( - func: FixtureFunction[C, P, R], - name: str = ..., - pattern: str = ..., -) -> FixtureFunction[C, P, R]: ... - -@overload -def fixture( - name: str = ..., - pattern: str = ..., -) -> FixtureDecorator: ... - - -class FixtureDecorator(Protocol): - - @overload - def __call__(self, _: FixtureCoroutine[C, P, R], /) -> FixtureCoroutine[C, P, R]: ... - - @overload - def __call__(self, _: FixtureFunction[C, P, R], /) -> FixtureFunction[C, P, R]: ... diff --git a/tests/stubs/behave/formatter/__init__.pyi b/tests/stubs/behave/formatter/__init__.pyi deleted file mode 100644 index e69de29bb2d1d6434b8b29ae775ad8c2e48c5391..0000000000000000000000000000000000000000 diff --git a/tests/stubs/behave/formatter/base.pyi b/tests/stubs/behave/formatter/base.pyi deleted file mode 100644 index df454b57618c2f4e204512c2f9259b2b675d4500..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/formatter/base.pyi +++ /dev/null @@ -1,58 +0,0 @@ -from typing import IO -from typing import ClassVar -from typing import Protocol - -from ..matchers import Match -from ..model import Background -from ..model import Configuration -from ..model import Feature -from ..model import Scenario -from ..model import Step - -class StreamOpener: - - default_encoding: ClassVar[str] - - name: str - stream: IO[str] - encoding: str - should_close_stream = bool - - def __init__( - self, - filename: str = ..., - stream: IO[str] = ..., - encoding: str = ..., - ): ... - - @staticmethod - def ensure_dir_exists(directory: str) -> None: ... - - @classmethod - def ensure_stream_with_encoder(cls, stream: IO[str], encoding: str = ...) -> IO[str]: ... - - def open(self) -> IO[str]: ... - def close(self) -> bool: ... - - -class Formatter(Protocol): - - name: str - description: str - - def __init__(self, stream_opener: StreamOpener, config: Configuration): ... - - @property - def stdout_mode(self) -> bool: ... - - def open(self) -> IO[str]: ... - def uri(self, uri: str) -> None: ... - def feature(self, feature: Feature) -> None: ... - def background(self, background: Background) -> None: ... - def scenario(self, scenario: Scenario) -> None: ... - def step(self, step: Step) -> None: ... - def match(self, match: Match) -> None: ... - def result(self, step: Step) -> None: ... - def eof(self) -> None: ... - def close(self) -> None: ... - def close_stream(self) -> None: ... diff --git a/tests/stubs/behave/log_capture.pyi b/tests/stubs/behave/log_capture.pyi deleted file mode 100644 index 4692f7ef4d29470281d0d589a2a419b06967036d..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/log_capture.pyi +++ /dev/null @@ -1,55 +0,0 @@ -from logging import Handler -from logging import LogRecord -from logging.handlers import BufferingHandler -from typing import Any -from typing import Callable -from typing import Protocol -from typing import TypeVar -from typing import overload - -from .model import Configuration -from .runner import ScenarioContext - -class RecordFilter: - - include: set[str] - exclude: set[str] - - def __init__(self, names: str): ... - - def filter(self, record: LogRecord) -> bool: ... - - -class LoggingCapture(BufferingHandler): - - config: Configuration - old_handlers: list[Handler] - old_level: int|None - - def __init__(self, config: Configuration, level: int = ...): ... - - def flush(self) -> None: ... - def truncate(self) -> None: ... - def getvalue(self) -> str: ... - def find_event(self, pattern: str) -> bool: ... - def any_errors(self) -> bool: ... - def inveigle(self) -> None: ... - def abandon(self) -> None: ... - - -MemoryHandler = LoggingCapture - - -class Hook(Protocol): - def __call__(self, _: ScenarioContext, /, *a: Any, **k: Any) -> None: ... - - -H = TypeVar("H", bound=Hook) - - -@overload -def capture(level: int = ...) -> Callable[[H], H]: ... - - -@overload -def capture(func: H, level: int = ...) -> H: ... diff --git a/tests/stubs/behave/matchers.pyi b/tests/stubs/behave/matchers.pyi deleted file mode 100644 index 8ff68f0644ca0951fece02da991afdfef58db3b0..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/matchers.pyi +++ /dev/null @@ -1,86 +0,0 @@ -from typing import Any -from typing import Callable -from typing import ClassVar -from typing import Optional -from typing import Sequence -from typing import TypeVar - -from parse import Parser -from parse import TypeConverter - -from .model_core import Argument -from .model_core import FileLocation -from .runner import Context -from .step_registry import StepFunction - -Arguments = Optional[list[Argument]] - -T = TypeVar("T") - - -matcher_mapping: dict[str, Matcher] -current_matcher: Matcher - - -def register_type(**kw: Callable[[str], Any]) -> None: ... -def use_step_matcher(name: str) -> None: ... -def get_matcher(func: StepFunction, pattern: str) -> Matcher: ... - - -class StepParserError(ValueError): ... - - -class Match: - - func: StepFunction - arguments: Sequence[Argument]|None - location: int - - def __init__(self, func: StepFunction, arguments: Sequence[Argument] = ...): ... - def with_arguments(self, arguments: Sequence[Argument]) -> Match: ... - def run(self, context: Context) -> None: ... - - @staticmethod - def make_location(step_function: StepFunction) -> FileLocation: ... - - -class NoMatch(Match): - - def __init__(self) -> None: ... - - -class MatchWithError(Match): - - def __init__(self, func: StepFunction, error: BaseException): ... - - -class Matcher: - - schema: ClassVar[str] - - pattern: str - func: StepFunction - - def __init__(self, func: StepFunction, pattern: str, step_type: str = ...): ... - - @property - def location(self) -> FileLocation: ... - - @property - def regex_pattern(self) -> str: ... - - def describe(self, schema: str = ...) -> str: ... - def check_match(self, step: str) -> Arguments: ... - def match(self, step: StepFunction) -> Match: ... - - -class ParseMatcher(Matcher): - - custom_types: ClassVar[dict[str, TypeConverter[Any]]] - parser_class: ClassVar[type[Parser]] - - -class CFParseMatcher(ParseMatcher): ... -class RegexMatcher(Matcher): ... -class SimplifiedRegexMatcher(RegexMatcher): ... -class CucumberRegexMatcher(RegexMatcher): ... diff --git a/tests/stubs/behave/model.pyi b/tests/stubs/behave/model.pyi deleted file mode 100644 index 9372d29b7090d351af6604d6fe1706f70fed73d2..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/model.pyi +++ /dev/null @@ -1,250 +0,0 @@ -from typing import Any -from typing import ClassVar -from typing import Iterable -from typing import Iterator -from typing import Literal -from typing import Protocol -from typing import Sequence - -from .model_core import BasicStatement -from .model_core import Replayable -from .model_core import Status -from .model_core import TagAndStatusStatement -from .runner import ModelRunner -from .tag_expression import TagExpression - -Configuration = dict[str, Any] - - -def reset_model(model_elements: Iterable[ModelElement]) -> None: ... - - -class ModelElement(Protocol): - - def reset(self) -> None: ... - - -class Feature(TagAndStatusStatement, Replayable): - - type: ClassVar[Literal["feature"]] - - keyword: str - name: str - description: list[str] - background: Background - scenarios: list[Scenario] - tags: list[Tag] - hook_failed: bool - filename: str - line: int - language: str - - @property - def status(self) -> Status: ... - - @property - def duration(self) -> float: ... - - def __init__( - self, - filename: str, - line: int, - keyword: str, - name: str, - tags: Sequence[Tag] = ..., - description: str = ..., - scenarios: Sequence[Scenario] = ..., - background: Background = ..., - language: str = ..., - ): ... - def __iter__(self) -> Iterator[Scenario]: ... - - def reset(self) -> None: ... - def add_scenario(self, scenario: Scenario) -> None: ... - def compute_status(self) -> Status: ... - def walk_scenarios(self, with_outlines: bool = ...) -> list[Scenario]: ... - def should_run(self, config: Configuration = ...) -> bool: ... - def should_run_with_tags(self, tag_expression: TagExpression) -> bool: ... - def mark_skipped(self) -> None: ... - def skip(self, reason: str = ..., require_not_executed: bool = ...) -> None: ... - def run(self, runner: ModelRunner) -> bool: ... - - -class Background(BasicStatement, Replayable): - - type: ClassVar[Literal["background"]] - - keyword: str - name: str - steps: list[Step] - filename: str - line: int - - @property - def duration(self) -> float: ... - - def __init__( - self, - filename: str, - line: int, - keyword: str, - name: str, - steps: Sequence[Step] = ..., - ): ... - def __iter__(self) -> Iterator[Step]: ... - - -class Scenario(TagAndStatusStatement, Replayable): - - type: ClassVar[Literal["scenario"]] - continue_after_failed_step: ClassVar[bool] - - keyword: str - name: str - description: str - feature: Feature - steps: Sequence[Step] - tags: Sequence[Tag] - hook_failed: bool - filename: str - line: int - - @property - def background_steps(self) -> list[Step]: ... - - @property - def all_steps(self) -> Iterator[Step]: ... - - @property - def duration(self) -> float: ... - - @property - def effective_tags(self) -> list[Tag]: ... - - def __init__( - self, - filename: str, - line: int, - keyword: str, - name: str, - tags: Sequence[Tag] = ..., - steps: Sequence[Step] = ..., - description: str = ..., - ): ... - def __iter__(self) -> Iterator[Step]: ... - - def reset(self) -> None: ... - def compute_status(self) -> Status: ... - def should_run(self, config: Configuration = ...) -> bool: ... - def should_run_with_tags(self, tag_expression: TagExpression) -> bool: ... - def should_run_with_name_select(self, config: Configuration) -> bool: ... - def mark_skipped(self) -> None: ... - def skip(self, reason: str = ..., require_not_executed: bool = ...) -> None: ... - def run(self, runner: ModelRunner) -> bool: ... - - -class Step(BasicStatement, Replayable): - - type: ClassVar[Literal["step"]] - - keyword: str - name: str - step_type: str - text: Text|None - table: Table|None - status: Status - hook_failed: bool - duration: float - error_message: str|None - filename: str - line: int - - def __init__( - self, - filename: str, - line: int, - keyword: str, - step_type: str, - name: str, - text: Text = ..., - table: Table = ..., - ): ... - def __eq__(self, other: Any) -> bool: ... - def __hash__(self) -> int: ... - - def reset(self) -> None: ... - def run(self, runner: ModelRunner, quiet: bool = ..., capture: bool = ...) -> bool: ... - - -class Table(Replayable): - - type: ClassVar[Literal["table"]] - - headings: Sequence[str] - line: int|None - rows: list[Row] - - def __init__(self, headings: Sequence[str], line: int = ..., rows: Sequence[Row] = ...): ... - def __eq__(self, other: Any) -> bool: ... - def __iter__(self) -> Iterator[Row]: ... - def __getitem__(self, index: int) -> Row: ... - - def add_row(self, row: Sequence[str], line: int) -> None: ... - def add_column(self, column_name: str, values: Iterable[str], default_value: str = ...) -> int: ... - def remove_column(self, column_name: str) -> None: ... - def remove_columns(self, column_names: Iterable[str]) -> None: ... - def has_column(self, column_name: str) -> bool: ... - def get_column_index(self, column_name: str) -> int: ... - def require_column(self, column_name: str) -> int: ... - def require_columns(self, column_names: Iterable[str]) -> None: ... - def ensure_column_exists(self, column_name: str) -> int: ... - def assert_equals(self, data: Table|Iterable[Row]) -> None: ... - - -class Row: - - headings: Sequence[str] - cells: Sequence[str] - line: int|None - comments: Sequence[str]|None - - def __init__( - self, - headings: Sequence[str], - cells: Sequence[str], - line: int = ..., - comments: Sequence[str] = ..., - ): ... - def __getitem__(self, index: int) -> str: ... - def __eq__(self, other: Any) -> bool: ... - def __len__(self) -> int: ... - def __iter__(self) -> Iterator[str]: ... - - def items(self) -> Iterator[tuple[str, str]]: ... - def get(self, key: int, default: str = ...) -> str: ... - def as_dict(self) -> dict[str, str]: ... - - -class Tag(str): - - allowed_chars: ClassVar[str] - quoting_chars: ClassVar[str] - - line: int - - def __init__(self, name: str, line: int): ... - - @classmethod - def make_name(cls, text: str, unexcape: bool = ..., allowed_chars: str = ...) -> str: ... - - -class Text(str): - - content_type: Literal["text/plain"] - line: int - - def __init__(self, value: str, content_type: Literal["text/plain"] = ..., line: int = ...): ... - - def line_range(self) -> tuple[int, int]: ... - def replace(self, old: str, new: str, count: int = ...) -> Text: ... - def assert_equals(self, expected: str) -> bool: ... diff --git a/tests/stubs/behave/model_core.pyi b/tests/stubs/behave/model_core.pyi deleted file mode 100644 index 44949165ce78a5bd04944542023cbfe1220f4e64..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/model_core.pyi +++ /dev/null @@ -1,123 +0,0 @@ -from enum import Enum -from types import TracebackType -from typing import Any -from typing import Callable -from typing import ClassVar -from typing import Sequence -from typing import TypeVar - -from .capture import Captured -from .formatter.base import Formatter -from .model import Tag -from .tag_expression import TagExpression - -class Status(Enum): - - untested = 0 - skipped = 1 - passed = 2 - failed = 3 - undefined = 4 - executing = 5 - - @classmethod - def from_name(cls, name: str) -> Status: ... - - -class Argument: - - original: str - value: Any - name: str|None - start: int - end: int - - def __init__(self, start: int, end: int, original: str, value: Any, name: str = ...): ... - - -class FileLocation: - - T = TypeVar("T", bound="FileLocation") - - filename: str - line: int - - def __init__(self, filename: str, line: int): ... - def __eq__(self, other: Any) -> bool: ... - def __ne__(self, other: Any) -> bool: ... - def __le__(self, other: str|FileLocation) -> bool: ... - def __gt__(self, other: str|FileLocation) -> bool: ... - def __ge__(self, other: str|FileLocation) -> bool: ... - def __str__(self) -> str: ... - - def get(self) -> str: ... - def abspath(self) -> str: ... - def basename(self) -> str: ... - def dirname(self) -> str: ... - def relpath(self, start: str = ...) -> str: ... - def exists(self) -> str: ... - - @classmethod - def for_function(cls: type[T], func: Callable[..., Any], curdir: str = ...) -> T: ... - - -class BasicStatement: - - location: FileLocation - keyword: str - name: str - captured: Captured - exception: Exception|None - exc_traceback: TracebackType|None - error_message: str|None - - @property - def filename(self) -> str: ... - - @property - def line(self) -> int: ... - - def __init__(self, filename: str, line: int, keyword: str, name: str): ... - def __hash__(self) -> int: ... - def __eq__(self, other: Any) -> bool: ... - def __ne__(self, other: Any) -> bool: ... - def __lt__(self, other: BasicStatement) -> bool: ... - def __le__(self, other: BasicStatement) -> bool: ... - def __gt__(self, other: BasicStatement) -> bool: ... - def __ge__(self, other: BasicStatement) -> bool: ... - - def reset(self) -> None: ... - def store_exception_context(self, exception: Exception) -> None: ... - - -class TagStatement(BasicStatement): - - tags: Sequence[Tag] - - def __init__(self, filename: str, line: int, keyword: str, name: str, tags: Sequence[Tag]): ... - def should_run_with_tags(self, tag_expression: TagExpression) -> bool: ... - - -class TagAndStatusStatement(BasicStatement): - - final_status: ClassVar[tuple[Status]] - - tags: Sequence[Tag] - should_skip: bool - skip_reason: str|None - - @property - def status(self) -> Status: ... - - def __init__(self, filename: str, line: int, keyword: str, name: str, tags: Sequence[Tag]): ... - - def should_run_with_tags(self, tag_expression: TagExpression) -> bool: ... - def set_status(self, value: Status) -> None: ... - def clear_status(self) -> None: ... - def reset(self) -> None: ... - def compute_status(self) -> Status: ... - - -class Replayable: - - def replay(self, formatter: Formatter) -> None: ... diff --git a/tests/stubs/behave/runner.pyi b/tests/stubs/behave/runner.pyi deleted file mode 100644 index 48deb102760d3929c1f1587c0e4146cd7b96d184..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/runner.pyi +++ /dev/null @@ -1,148 +0,0 @@ -import contextlib -from io import StringIO -from typing import Any -from typing import Callable -from typing import Iterator -from typing import Literal -from typing import Protocol -from typing import Sequence -from typing import Union - -from .capture import CaptureController -from .formatter.base import Formatter -from .log_capture import LoggingCapture -from .model import Configuration -from .model import Feature -from .model import Row -from .model import Scenario -from .model import Step -from .model import Table -from .model import Tag -from .model_core import FileLocation -from .step_registry import StepRegistry - -Mode = Union[Literal["behave"], Literal["user"]] - - -@contextlib.contextmanager -def use_context_with_mode(context: Context, mode: Mode) -> Iterator[None]: ... - -@contextlib.contextmanager -def scoped_context_layer(context: Context, layer_name: str|None = None) -> Iterator[Context]: ... - -def path_getrootdir(path: str)-> str: ... - - -class CleanupError(RuntimeError): ... -class ContextMaskWarning(UserWarning): ... - - -class Context(Protocol): - - def __getattr__(self, name: str) -> Any: ... - def __setattr__(self, name: str, value: Any) -> None: ... - def __contains__(self, name: str) -> bool: ... - - def add_cleanup(self, cleanup_func: Callable[..., None], *a: Any, **k: Any) -> None: ... - - @property - def config(self) -> Configuration: ... - - @property - def aborted(self) -> bool: ... - - @property - def failed(self) -> bool: ... - - @property - def log_capture(self) -> LoggingCapture|None: ... - - @property - def stdout_capture(self) -> StringIO|None: ... - - @property - def stderr_capture(self) -> StringIO|None: ... - - @property - def cleanup_errors(self) -> int: ... - - # Feature always present, None outside of feature namespace - - @property - def feature(self) -> Feature|None: ... - - # Step values always present, may be None even in step namespace - - @property - def active_outline(self) -> Row|None: ... - - @property - def table(self) -> Table|None: ... - - @property - def text(self) -> str|None: ... - - -class FeatureContext(Protocol, Context): - - def execute_steps(self, steps_text: str) -> bool: ... - - @property - def feature(self) -> Feature: ... - - @property - def tags(self) -> set[Tag]: ... - - -class ScenarioContext(Protocol, FeatureContext): - - @property - def scenario(self) -> Scenario: ... - - -class Hook(Protocol): - - def __call__(self, context: Context, *args: Any) -> None: ... - - -class ModelRunner: - - config: Configuration - features: Sequence[Feature] - step_registry: StepRegistry - hooks: dict[str, Hook] - formatters: list[Formatter] - undefined_steps: list[Step] - capture_controller: CaptureController - context: Context|None - feature: Feature|None - hook_failures: int - - # is a property in concrete class - aborted: bool - - def __init__( - self, - config: Configuration, - features: Sequence[Feature]|None, - step_registry: StepRegistry|None, - ): ... - - def run_hook(self, name: str, context: Context, *args: Any) -> None: ... - def setup_capture(self) -> None: ... - def start_capture(self) -> None: ... - def stop_capture(self) -> None: ... - def teardown_capture(self) -> None: ... - def run_model(self, features: Sequence[Feature]|None) -> bool: ... - def run(self) -> bool: ... - - -class Runner(ModelRunner): - - def __init__(self, config: Configuration): ... - def setup_paths(self) -> None: ... - def before_all_default_hook(self, context: Context) -> None: ... - def load_hooks(self, filename: str = ...) -> None: ... - def load_step_definitions(self, extra_step_paths: Sequence[str] = ...) -> None: ... - def feature_locations(self) -> list[FileLocation]: ... - def run_with_paths(self) -> bool: ... diff --git a/tests/stubs/behave/step_registry.pyi b/tests/stubs/behave/step_registry.pyi deleted file mode 100644 index 8609d1ceef0647905a22cd58e8ffa7ff7e2bf7ff..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/step_registry.pyi +++ /dev/null @@ -1,27 +0,0 @@ -from typing import Callable -from typing import TypeVar - -C = TypeVar("C") -StepDecorator = Callable[[str], Callable[[C], C]] -StepFunction = Callable[..., None] - - -Given: StepDecorator[StepFunction] -given: StepDecorator[StepFunction] - -When: StepDecorator[StepFunction] -when: StepDecorator[StepFunction] - -Then: StepDecorator[StepFunction] -then: StepDecorator[StepFunction] - -Step: StepDecorator[StepFunction] -step: StepDecorator[StepFunction] - - -class AmbiguousStep(ValueError): ... - - -class StepRegistry: - - steps: dict[str, list[StepFunction]] diff --git a/tests/stubs/behave/tag_expression.pyi b/tests/stubs/behave/tag_expression.pyi deleted file mode 100644 index cdb025b366fcc2bd780376f604bf42aa33be1c50..0000000000000000000000000000000000000000 --- a/tests/stubs/behave/tag_expression.pyi +++ /dev/null @@ -1,22 +0,0 @@ -from typing import Iterable -from typing import Sequence - -from .model import Tag - -class TagExpression: - - ands: list[tuple[str]] - limits: dict[tuple[str], int] - - def __init__(self, tag_expressions: Iterable[str]): ... - def __len__(self) -> int: ... - - def check(self, tags: Sequence[Tag]) -> bool: ... - - @staticmethod - def normalize_tag(tag: str) -> str: ... - - @classmethod - def normalized_tags_from_or(cls, expr: str) -> Iterable[str]: ... - - def store_and_extract_limits(self, tags: Iterable[str]) -> None: ... diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py deleted file mode 100644 index 8c8fde56fa08ba7d68b44acf0c17d4fa131d9b6e..0000000000000000000000000000000000000000 --- a/tests/utils/__init__.py +++ /dev/null @@ -1,29 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -A toolkit of helpful functions and classes for step implementations -""" - -from __future__ import annotations - -from .behave import PatternEnum -from .behave import register_pattern -from .http import redirect -from .json import JSONArray -from .json import JSONObject -from .secret import make_secret -from .url import URL - -__all__ = ( - "JSONArray", - "JSONObject", - "PatternEnum", - "URL", - "make_secret", - "redirect", - "register_pattern", -) diff --git a/tests/utils/behave.py b/tests/utils/behave.py deleted file mode 100644 index 765b62abb92b5945559d6a2b5000224af599d28a..0000000000000000000000000000000000000000 --- a/tests/utils/behave.py +++ /dev/null @@ -1,115 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -Utilities for "behave" interactions -""" - -from __future__ import annotations - -import enum -from typing import TYPE_CHECKING -from typing import Any -from typing import Protocol -from typing import TypeVar -from typing import overload - -import behave -import parse - -T = TypeVar("T") - -__all__ = [ - "PatternEnum", - "register_pattern", -] - - -class PatternConverter(Protocol): - - __name__: str - - def __call__(self, match: str) -> Any: ... - - -class Decorator(Protocol[T]): - - def __call__(self, converter: T) -> T: ... - - -@overload -def register_pattern(pattern: str) -> Decorator[PatternConverter]: ... - - -@overload -def register_pattern( - pattern: str, - converter: PatternConverter, - name: str = ..., -) -> PatternConverter: ... - - -def register_pattern( - pattern: str, - converter: PatternConverter|None = None, - name: str = "", -) -> PatternConverter|Decorator[PatternConverter]: - """ - Register a pattern and converter for a step parser type - - The type is named after the converter. - """ - pattern_decorator = parse.with_pattern(pattern) - - def decorator(converter: PatternConverter) -> PatternConverter: - nonlocal name - name = name or converter.__name__ - behave.register_type(**{name: pattern_decorator(converter)}) - return converter - - if converter: - return decorator(converter) - return decorator - - -class EnumMeta(enum.EnumMeta): - - MEMBER_FILTER = 'member_filter' - - T = TypeVar("T", bound="EnumMeta") - - def __new__(mtc: type[T], name: str, bases: tuple[type, ...], attr: dict[str, Any], **kwds: Any) -> T: - member_names: list[str] = attr._member_names # type: ignore - member_filter = attr.pop(mtc.MEMBER_FILTER, None) - if member_filter: - assert isinstance(member_filter, staticmethod) - member_filter.__func__(attr, member_names) - cls = enum.EnumMeta.__new__(mtc, name, bases, attr, **kwds) - decorator = parse.with_pattern('|'.join(member for member in cls.__members__)) - behave.register_type(**{name: decorator(cls)}) - return cls - - -class PatternEnum(enum.Enum, metaclass=EnumMeta): - """ - An enum class that self registers as a pattern type for step implementations - - Enum names are used to match values in step texts, so a value can be aliased multiple - times to provide alternates for matching, including alternative languages. - To supply names that are not valid identifiers the functional Enum API must be used, - supplying mapped values: - https://docs.python.org/3/library/enum.html#functional-api - - Enum values may be anything meaningful; for instance a command keyword that identifies a - type. - """ - - if TYPE_CHECKING: - C = TypeVar("C", bound="PatternEnum") - - @classmethod - def _missing_(cls: type[C], key: Any) -> C: - return cls[key] diff --git a/tests/utils/http.py b/tests/utils/http.py deleted file mode 100644 index b88484c777c6e4329bfcb25213a8c22d0f044569..0000000000000000000000000000000000000000 --- a/tests/utils/http.py +++ /dev/null @@ -1,70 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -Extensions for "requests" -""" - -from __future__ import annotations - -import ipaddress -from typing import Any -from typing import Mapping -from urllib.parse import urlparse - -import requests.adapters -from requests.packages.urllib3 import connection -from requests.packages.urllib3 import connectionpool - - -def redirect(session: requests.Session, prefix: str, address: ipaddress.IPv4Address) -> None: - """ - Redirect all requests for "prefix" to a given address - - This function allows a user to completely override DNS and local name lookups, allowing - fixtures to be contacted via any configured URL without having to mess with the system's - name resolution services. - - "prefix" is formated as either "{hostname}[:{port}]" or "{schema}://{hostname}[:{port}]" - where "schema" defaults to (and currently only supports) "http". - """ - if prefix.startswith("https://"): - raise ValueError("https:// prefixes not currently supported") - if not prefix.startswith("http://"): - prefix = f"http://{prefix}" - session.mount(prefix, LocalHTTPAdapter(address)) - - -class LocalHTTPAdapter(requests.adapters.HTTPAdapter): - """ - An alternative HTTP adapter that directs all connections to a configured address - - Instances of this class are mounted on a `requests.Session` as adapters for specific URL - prefixes. - - Rather than using this class directly the easiest way to use it is with the `redirect` - function. - """ - - def __init__(self, destination: ipaddress.IPv4Address): - super().__init__() - self.destination = destination - - def get_connection(self, url: str, proxies: Mapping[str, str]|None = None) -> _HTTPConnectionPool: - parts = urlparse(url) - return _HTTPConnectionPool(parts.hostname, parts.port, address=self.destination) - - -class _HTTPConnectionPool(connectionpool.HTTPConnectionPool): - - class ConnectionCls(connection.HTTPConnection): - - # Undo the damage done by parent class which makes 'host' a property with magic - host = "" - - def __init__(self, /, address: ipaddress.IPv4Address, **kwargs: Any): - connection.HTTPConnection.__init__(self, **kwargs) - self._dns_host = str(address) diff --git a/tests/utils/json.py b/tests/utils/json.py deleted file mode 100644 index fd9d43bba0fc33bdbaab5930e1511c8c8743aa15..0000000000000000000000000000000000000000 --- a/tests/utils/json.py +++ /dev/null @@ -1,68 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -JSON classes for container types (objects and arrays) -""" - -from __future__ import annotations - -import json -from typing import Any -from typing import Callable -from typing import TypeVar -from typing import overload - -from jsonpath import JSONPath - -__all__ = [ - "JSONObject", - "JSONArray", -] - - -class JSONPathMixin: - - T = TypeVar('T', bound=object) - C = TypeVar('C', bound=object) - - @overload - def path(self, path: str, kind: type[T], convert: None = None) -> T: ... - - @overload - def path(self, path: str, kind: type[T], convert: Callable[[T], C]) -> C: ... - - def path(self, path: str, kind: type[T], convert: Callable[[T], C]|None = None) -> T|C: - result = JSONPath(path).parse(self)[0] - if convert is not None: - return convert(result) - elif isinstance(result, kind): - return result - raise ValueError(f"{path} is wrong type; expected {kind}; got {type(result)}") - - -class JSONObject(JSONPathMixin, dict[str, Any]): - """ - A dict for JSON objects that implements `.path` for getting child items by a JSON path - """ - - T = TypeVar("T", bound="JSONObject") - - @classmethod - def from_string(cls: type[T], string: bytes) -> T: - return cls(json.loads(string)) - - -class JSONArray(JSONPathMixin, list[Any]): - """ - A list for JSON arrays that implements `.path` for getting child items by a JSON path - """ - - T = TypeVar("T", bound="JSONArray") - - @classmethod - def from_string(cls: type[T], string: bytes) -> T: - return cls(json.loads(string)) diff --git a/tests/utils/secret.py b/tests/utils/secret.py deleted file mode 100644 index b5caa58bb2c1c34bf83ce360dccdd3120556b166..0000000000000000000000000000000000000000 --- a/tests/utils/secret.py +++ /dev/null @@ -1,23 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -Utility for generating secrets -""" - -from __future__ import annotations - -import string -from secrets import choice - -CHARS = string.ascii_letters + string.digits - - -def make_secret(size: int) -> str: - """ - Generate a string of alphanumeric characters for use as a password - """ - return ''.join(choice(CHARS) for _ in range(size)) diff --git a/tests/utils/url.py b/tests/utils/url.py deleted file mode 100644 index 2fc6f9bd301297aff4a9a2588ae59c18c33bbd8e..0000000000000000000000000000000000000000 --- a/tests/utils/url.py +++ /dev/null @@ -1,28 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -URL types and pattern matcher -""" - -from __future__ import annotations - -from urllib.parse import urljoin - -from .behave import register_pattern - - -@register_pattern(r"(?:https?://\S+|/\S*)") -class URL(str): - """ - A subclass for URL strings which also acts as a pattern match type - """ - - def __truediv__(self, other: str) -> URL: - return URL(urljoin(self, other)) - - def __add__(self, other: str) -> URL: - return URL(str(self) + other) diff --git a/tests/wp.py b/tests/wp.py new file mode 100644 index 0000000000000000000000000000000000000000..9d22f216c59c1609e7810164d58dcb7fde2e3c8f --- /dev/null +++ b/tests/wp.py @@ -0,0 +1,126 @@ +# Copyright 2021 Dominik Sekotill +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +""" +Management and control for WordPress fixtures +""" + +from __future__ import annotations + +from contextlib import contextmanager +from os import environ +from pathlib import Path +from typing import Iterator +from typing import NamedTuple + +from behave_utils import URL +from behave_utils import wait +from behave_utils.docker import Cli +from behave_utils.docker import Container as Container +from behave_utils.docker import Image +from behave_utils.docker import IPv4Address +from behave_utils.docker import Network +from behave_utils.mysql import Mysql + +BUILD_CONTEXT = Path(__file__).parent.parent + + +class Wordpress(Container): + """ + Container subclass for a WordPress PHP-FPM container + """ + + DEFAULT_ALIASES = ("upstream",) + + def __init__(self, site_url: URL, database: Mysql, network: Network|None = None): + Container.__init__( + self, + Image.build( + BUILD_CONTEXT, + php_version=environ.get("PHP_VERSION"), + wp_version=environ.get("WP_VERSION"), + ), + volumes=[ + ("static", Path("/app/static")), + ("media", Path("/app/media")), + ], + env=dict( + SITE_URL=site_url, + SITE_ADMIN_EMAIL="test@kodo.org.uk", + DB_NAME=database.name, + DB_USER=database.user, + DB_PASS=database.password, + DB_HOST=database.get_location(), + ), + network=network, + ) + + @property + def cli(self) -> Cli: + """ + Run WP-CLI commands + """ + return Cli(self, "wp") + + @contextmanager + def started(self) -> Iterator[Container]: + with self: + self.start() + cmd = ["bash", "-c", "[[ /proc/1/exe -ef `which php-fpm` ]]"] + wait(lambda: self.is_running() and self.run(cmd).returncode == 0, timeout=600) + yield self + + +class Nginx(Container): + """ + Container subclass for an Nginx frontend + """ + + def __init__(self, backend: Wordpress, network: Network|None = None): + Container.__init__( + self, + Image.build( + BUILD_CONTEXT, + target='nginx', + nginx_version=environ.get("NGINX_VERSION"), + ), + network=network, + volumes=backend.volumes, + ) + + +class Site(NamedTuple): + """ + A named-tuple of information about the containers for a site fixture + """ + + url: str + address: IPv4Address + frontend: Nginx + backend: Wordpress + database: Mysql + + +@contextmanager +def test_cluster(site_url: URL) -> Iterator[Site]: + """ + Configure and start all the necessary containers for use as test fixtures + """ + test_dir = Path(__file__).parent + db_init = test_dir / "mysql-init.sql" + + with Network() as network: + database = Mysql(network=network, init_files=[db_init]) + database.start() # Get a head start on initialising the database + backend = Wordpress(site_url, database, network=network) + frontend = Nginx(backend, network=network) + + with database.started(), backend.started(), frontend.started(): + addr = frontend.inspect().path( + f"$.NetworkSettings.Networks.{network}.IPAddress", + str, IPv4Address, + ) + yield Site(site_url, addr, frontend, backend, database) diff --git a/tests/wp/__init__.py b/tests/wp/__init__.py deleted file mode 100644 index 8a6f487b957b9006e1f39cf79baa0e8ecbfa6141..0000000000000000000000000000000000000000 --- a/tests/wp/__init__.py +++ /dev/null @@ -1,177 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -Management and control for WordPress fixtures -""" - -from __future__ import annotations - -from contextlib import contextmanager -from subprocess import CalledProcessError -from time import sleep -from time import time -from typing import Any -from typing import Callable -from typing import Iterator -from typing import Literal -from typing import SupportsBytes -from typing import TypeVar -from typing import overload - -from .docker import Container as Container -from .proc import PathArg -from .proc import coerce_args -from .proc import exec_io - - -def wait(predicate: Callable[[], bool], timeout: float = 120.0) -> None: - """ - Block and periodically call "predictate" until it returns True, or the time limit passes - """ - end = time() + timeout - left = timeout - while left > 0.0: - sleep( - 10 if left > 60.0 else - 5 if left > 10.0 else - 1, - ) - left = end - time() - if predicate(): - return - raise TimeoutError - - -class Cli: - """ - Manage calling executables in a container - - Any arguments passed to the constructor will prefix the arguments passed when the object - is called. - """ - - T = TypeVar("T") - - def __init__(self, container: Container, *cmd: PathArg): - self.container = container - self.cmd = cmd - - @overload - def __call__( - self, - *args: PathArg, - input: str|SupportsBytes|None = ..., - deserialiser: Callable[[bytes], T], - query: Literal[False], - **kwargs: Any, - ) -> T: ... - - @overload - def __call__( - self, - *args: PathArg, - input: str|SupportsBytes|None = ..., - deserialiser: None = None, - query: Literal[True], - **kwargs: Any, - ) -> int: ... - - @overload - def __call__( - self, - *args: PathArg, - input: str|SupportsBytes|None = ..., - deserialiser: None = None, - query: Literal[False], - **kwargs: Any, - ) -> None: ... - - def __call__( - self, - *args: PathArg, - input: str|SupportsBytes|None = None, - deserialiser: Callable[[bytes], T]|None = None, - query: bool = False, - **kwargs: Any, - ) -> Any: - # deserialiser = kwargs.pop('deserialiser', None) - assert not deserialiser or not query - - data = ( - b"" if input is None else - input.encode() if isinstance(input, str) else - bytes(input) - ) - cmd = self.container.get_exec_args([*self.cmd, *args], interactive=bool(data)) - - if deserialiser: - return exec_io(cmd, data, deserialiser=deserialiser, **kwargs) - - rcode = exec_io(cmd, data, **kwargs) - if query: - return rcode - if not isinstance(rcode, int): - raise TypeError(f"got rcode {rcode!r}") - if 0 != rcode: - raise CalledProcessError(rcode, ' '.join(coerce_args(cmd))) - return None - - -class Wordpress(Container): - """ - Container subclass for a WordPress PHP-FPM container - """ - - @property - def cli(self) -> Cli: - """ - Run WP-CLI commands - """ - return Cli(self, "wp") - - @contextmanager - def started(self) -> Iterator[Container]: - with self: - self.start() - cmd = ["bash", "-c", "[[ /proc/1/exe -ef `which php-fpm` ]]"] - wait(lambda: self.is_running() and self.run(cmd).returncode == 0, timeout=600) - yield self - - -class Mysql(Container): - """ - Container subclass for a database container - """ - - @property - def mysql(self) -> Cli: - """ - Run "mysql" commands - """ - return Cli(self, "mysql") - - @property - def mysqladmin(self) -> Cli: - """ - Run "mysqladmin" commands - """ - return Cli(self, "mysqladmin") - - @property - def mysqldump(self) -> Cli: - """ - Run "mysqldump" commands - """ - return Cli(self, "mysqldump") - - @contextmanager - def started(self) -> Iterator[Container]: - with self: - self.start() - sleep(20) - wait(lambda: self.run(['/healthcheck.sh']).returncode == 0) - yield self diff --git a/tests/wp/docker.py b/tests/wp/docker.py deleted file mode 100644 index 9a28d3ceccc258affcf67a6dab964a97bd804f38..0000000000000000000000000000000000000000 --- a/tests/wp/docker.py +++ /dev/null @@ -1,414 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -Commands for managing Docker for fixtures -""" - -from __future__ import annotations - -import ipaddress -import json -from contextlib import contextmanager -from pathlib import Path -from secrets import token_hex -from subprocess import DEVNULL -from subprocess import PIPE -from subprocess import CompletedProcess -from subprocess import Popen -from subprocess import run -from types import TracebackType -from typing import IO -from typing import Any -from typing import Callable -from typing import Iterable -from typing import Iterator -from typing import Optional -from typing import Tuple -from typing import TypeVar -from typing import Union -from typing import overload - -from jsonpath import JSONPath - -from .proc import Arguments -from .proc import Environ -from .proc import MutableArguments -from .proc import PathArg -from .proc import PathLike -from .proc import coerce_args - -T_co = TypeVar('T_co', covariant=True) - -HostMount = tuple[PathLike, PathLike] -NamedMount = tuple[str, PathLike] -AnonMount = PathLike -Mount = Union[HostMount, NamedMount, AnonMount] -Volumes = Iterable[Mount] - -DOCKER = 'docker' - - -def docker(*args: PathArg, **env: str) -> None: - """ - Run a Docker command, with output going to stdout - """ - run([DOCKER, *coerce_args(args)], env=env, check=True) - - -def docker_output(*args: PathArg, **env: str) -> str: - """ - Run a Docker command, capturing and returning its stdout - """ - proc = run([DOCKER, *coerce_args(args)], env=env, check=True, stdout=PIPE, text=True) - return proc.stdout.strip() - - -def docker_quiet(*args: PathArg, **env: str) -> None: - """ - Run a Docker command, directing its stdout to /dev/null - """ - run([DOCKER, *coerce_args(args)], env=env, check=True, stdout=DEVNULL) - - -class IPv4Address(ipaddress.IPv4Address): - """ - Subclass of IPv4Address that handle's docker idiosyncratic tendency to add a mask suffix - """ - - T = TypeVar("T", bound="IPv4Address") - - @classmethod - def with_suffix(cls: type[T], address: str) -> T: - """ - Construct an instance with a suffixed bitmask size - """ - address, *_ = address.partition("/") - return cls(address) - - -class Item: - """ - A mix-in for Docker items that can be inspected - """ - - T = TypeVar('T', bound=object) - C = TypeVar('C', bound=object) - - _data: Optional[dict[str, Any]] = None - - def __init__(self, name: str): - self.name = name - self._data: Any = None - - def get_id(self) -> str: - """ - Return an identifier for the Docker item - """ - return self.name - - @overload - def inspect(self, path: str, kind: type[T], convert: None = None) -> T: ... - - @overload - def inspect(self, path: str, kind: type[T], convert: Callable[[T], C]) -> C: ... - - def inspect(self, path: str, kind: type[T], convert: Callable[[T], C]|None = None) -> T|C: - """ - Extract a value from an item's information by JSON path - - "kind" is the type of the extracted value, while "convert" is an optional callable - that can turn that type to another type. The return type will be the return type of - "convert" if provided, "kind" otherwise. - """ - if self._data is None: - with Popen([DOCKER, 'inspect', self.get_id()], stdout=PIPE) as proc: - assert proc.stdout is not None - results = json.load(proc.stdout) - assert isinstance(results, list) - assert len(results) == 1 and isinstance(results[0], dict) - self._data = results[0] - result = JSONPath(path).parse(self._data) - if "*" not in path: - try: - result = result[0] - except IndexError: - raise KeyError(path) from None - if not isinstance(result, kind): - raise TypeError(f"{path} is wrong type; expected {kind}; got {type(result)}") - if convert is None: - return result - return convert(result) - - -class Image(Item): - """ - Docker image items - """ - - T = TypeVar('T', bound='Image') - - def __init__(self, iid: str): - self.iid = iid - - @classmethod - def build(cls: type[T], context: Path, target: str = "", **build_args: str|None) -> T: - """ - Build an image from the given context - - Build arguments are ignored if they are None to make it easier to supply (or not) - arguments from external lookups without complex argument composing. - """ - cmd: Arguments = [ - 'build', context, f"--target={target}", - *(f"--build-arg={arg}={val}" for arg, val in build_args.items() if val is not None), - ] - docker(*cmd, DOCKER_BUILDKIT='1') - iid = docker_output(*cmd, '-q', DOCKER_BUILDKIT='1') - return cls(iid) - - @classmethod - def pull(cls: type[T], repository: str) -> T: - """ - Pull an image from a registry - """ - docker('pull', repository) - iid = Item(repository).inspect('$.Id', str) - return cls(iid) - - def get_id(self) -> str: - return self.iid - - -class Container(Item): - """ - Docker container items - - Instances can be used as context managers that ensure the container is stopped on - exiting the context. - """ - - DEFAULT_ALIASES = tuple[str]() - - def __init__( - self, - image: Image, - cmd: Arguments = [], - volumes: Volumes = [], - env: Environ = {}, - network: Network|None = None, - entrypoint: HostMount|PathArg|None = None, - ): - if isinstance(entrypoint, tuple): - volumes = [*volumes, entrypoint] - entrypoint = entrypoint[1] - - self.image = image - self.cmd = cmd - self.volumes = volumes - self.env = env - self.entrypoint = entrypoint - self.networks = dict[Network, Tuple[str, ...]]() - self._id: str|None = None - - if network: - self.networks[network] = Container.DEFAULT_ALIASES - - def __enter__(self) -> Container: - return self - - def __exit__(self, etype: type[BaseException], exc: BaseException, tb: TracebackType) -> None: - if self._id and exc: - self.show_logs() - self.stop(rm=True) - - @contextmanager - def started(self) -> Iterator[Container]: - """ - A context manager that ensures the container is started when the context is entered - """ - with self: - self.start() - yield self - - def is_running(self) -> bool: - """ - Return whether the container is running - """ - if self._id is None: - return False - item = Item(self._id) - if item.inspect('$.State.Status', str) == 'exited': - code = item.inspect('$.State.ExitCode', int) - raise ProcessLookupError(f"container {self._id} exited ({code})") - return ( - self._id is not None - and item.inspect('$.State.Running', bool) - ) - - def get_id(self) -> str: - if self._id is not None: - return self._id - - networks = set[Network]() - opts: MutableArguments = [ - *( - (f"--volume={vol[0]}:{vol[1]}" if isinstance(vol, tuple) else f"--volume={vol}") - for vol in self.volumes - ), - *(f"--env={name}={val}" for name, val in self.env.items()), - ] - - if self.entrypoint: - opts.append(f"--entrypoint={self.entrypoint}") - if self.networks: - networks.update(self.networks) - net = networks.pop() - opts.append(f"--network={net}") - opts.extend(f"--network-alias={alias}" for alias in self.networks[net]) - - self._id = docker_output('container', 'create', *opts, self.image.iid, *self.cmd) - assert self._id - return self._id - - def start(self) -> None: - """ - Start the container - """ - docker_quiet('container', 'start', self.get_id()) - - def stop(self, rm: bool = False) -> None: - """ - Stop the container - """ - if self._id is None: - return - docker_quiet('container', 'stop', self._id) - if rm: - docker_quiet('container', 'rm', self._id) - self._id = None - - def connect(self, network: Network, *aliases: str) -> None: - """ - Connect the container to a Docker network - - Any aliases supplied will be resolvable to the container by other containers on the - network. - """ - is_running = self.is_running() - if network in self.networks: - if self.networks[network] == aliases: - return - if is_running: - docker('network', 'disconnect', str(network), self.get_id()) - if is_running: - docker( - 'network', 'connect', - *(f'--alias={a}' for a in aliases), - str(network), self.get_id(), - ) - self.networks[network] = aliases - - def show_logs(self) -> None: - """ - Print the container logs to stdout - """ - if self._id: - docker('logs', self._id) - - def get_exec_args(self, cmd: Arguments, interactive: bool = False) -> MutableArguments: - """ - Return a full argument list for running "cmd" inside the container - """ - return [DOCKER, "exec", *(("-i",) if interactive else ""), self.get_id(), *coerce_args(cmd)] - - def run( - self, - cmd: Arguments, - *, - stdin: IO[Any]|int|None = None, - stdout: IO[Any]|int|None = None, - stderr: IO[Any]|int|None = None, - capture_output: bool = False, - check: bool = False, - input: bytes|None = None, - timeout: float|None = None, - ) -> CompletedProcess[bytes]: - """ - Run "cmd" to completion inside the container and return the result - """ - return run( - self.get_exec_args(cmd), - stdin=stdin, stdout=stdout, stderr=stderr, - capture_output=capture_output, - check=check, timeout=timeout, input=input, - ) - - def exec( - self, - cmd: Arguments, - *, - stdin: IO[Any]|int|None = None, - stdout: IO[Any]|int|None = None, - stderr: IO[Any]|int|None = None, - ) -> Popen[bytes]: - """ - Execute "cmd" inside the container and return a process object once started - """ - return Popen( - self.get_exec_args(cmd), - stdin=stdin, stdout=stdout, stderr=stderr, - ) - - -class Network: - """ - A Docker network - """ - - def __init__(self, name: str|None = None) -> None: - self._name = name or f"br{token_hex(6)}" - - def __str__(self) -> str: - return self._name - - def __repr__(self) -> str: - cls = type(self) - return f"<{cls.__module__}.{cls.__name__} {self._name}>" - - def __eq__(self, other: Any) -> bool: - if not isinstance(other, Network): - return self._name == str(other) - return self._name == other._name - - def __hash__(self) -> int: - return self._name.__hash__() - - def __enter__(self) -> Network: - self.create() - return self - - def __exit__(self, etype: type[BaseException], exc: BaseException, tb: TracebackType) -> None: - self.destroy() - - @property - def name(self) -> str: - return self._name - - def get_id(self) -> str: - return self._name - - def create(self) -> None: - """ - Create the network - """ - docker_quiet("network", "create", self._name) - - def destroy(self) -> None: - """ - Remove the network - """ - docker_quiet("network", "rm", self._name) diff --git a/tests/wp/proc.py b/tests/wp/proc.py deleted file mode 100644 index e368af68a13cf1428a3e62a5afca37b9229dc8c5..0000000000000000000000000000000000000000 --- a/tests/wp/proc.py +++ /dev/null @@ -1,158 +0,0 @@ -# Copyright 2021 Dominik Sekotill -# -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. - -""" -Manage processes asynchronously -""" - -from __future__ import annotations - -import io -import logging -import os -import sys -from subprocess import DEVNULL -from subprocess import PIPE -from typing import IO -from typing import Any -from typing import Callable -from typing import Iterator -from typing import Mapping -from typing import MutableSequence -from typing import Sequence -from typing import TypeVar -from typing import Union -from typing import overload - -import trio.abc - -T = TypeVar('T') -Deserialiser = Callable[[bytes], T] - -PathLike = os.PathLike[str] -PathArg = Union[PathLike, str] -Arguments = Sequence[PathArg] -MutableArguments = MutableSequence[PathArg] -Environ = Mapping[str, str] - -_logger = logging.getLogger(__name__) - - -def coerce_args(args: Arguments) -> Iterator[str]: - """ - Ensure path-like arguments are converted to strings - """ - return (os.fspath(a) for a in args) - - -@overload -def exec_io( - cmd: Arguments, - data: bytes = b'', - deserialiser: Deserialiser[T] = ..., - **kwargs: Any, -) -> T: ... - - -@overload -def exec_io( - cmd: Arguments, - data: bytes = b'', - deserialiser: None = None, - **kwargs: Any, -) -> int: ... - - -def exec_io( - cmd: Arguments, - data: bytes = b'', - deserialiser: Deserialiser[Any]|None = None, - **kwargs: Any, -) -> Any: - """ - Execute a command, handling output asynchronously - - If data is provided it will be fed to the process' stdin. - If a deserialiser is provided it will be used to parse stdout data from the process. - - Stderr and stdout (if no deserialiser is provided) will be written to `sys.stderr` and - `sys.stdout` respectively. - - Note that the data is written, not redirected. If either `sys.stdout` or `sys.stderr` - is changed to an IO-like object with no file descriptor, this will still work. - """ - if deserialiser and 'stdout' in kwargs: - raise TypeError("Cannot provide 'deserialiser' with 'stdout' argument") - if data and 'stdin' in kwargs: - raise TypeError("Cannot provide 'data' with 'stdin' argument") - stdout: IO[str]|IO[bytes]|int = io.BytesIO() if deserialiser else kwargs.pop('stdout', sys.stdout) - stderr: IO[str]|IO[bytes]|int = kwargs.pop('stderr', sys.stderr) - _logger.debug("executing: %s", cmd) - proc = trio.run(_exec_io, cmd, data, stdout, stderr, kwargs) - if deserialiser: - assert isinstance(stdout, io.BytesIO) - return deserialiser(stdout.getvalue()) - return proc.returncode - - -async def _exec_io( - cmd: Arguments, - data: bytes, - stdout: IO[str]|IO[bytes]|int, - stderr: IO[str]|IO[bytes]|int, - kwargs: dict[str, Any], -) -> trio.Process: - proc = await trio.open_process( - [*coerce_args(cmd)], - stdin=PIPE if data else DEVNULL, - stdout=PIPE, - stderr=PIPE, - **kwargs, - ) - async with proc, trio.open_nursery() as nursery: - assert proc.stdout is not None and proc.stderr is not None - nursery.start_soon(_passthru, proc.stderr, stderr) - nursery.start_soon(_passthru, proc.stdout, stdout) - if data: - assert proc.stdin is not None - async with proc.stdin as stdin: - await stdin.send_all(data) - return proc - - -async def _passthru(in_stream: trio.abc.ReceiveStream, out_stream: IO[str]|IO[bytes]|int) -> None: - try: - if not isinstance(out_stream, int): - out_stream = out_stream.fileno() - except (OSError, AttributeError): - # cannot get file descriptor, probably a memory buffer - if isinstance(out_stream, io.BytesIO): - async def write(data: bytes) -> None: - assert isinstance(out_stream, io.BytesIO) - out_stream.write(data) - elif isinstance(out_stream, io.StringIO): - async def write(data: bytes) -> None: - assert isinstance(out_stream, io.StringIO) - out_stream.write(data.decode()) - else: - raise TypeError(f"Unknown IO type: {type(out_stream)}") - else: - # is/has a file descriptor, out_stream is now that file descriptor - async def write(data: bytes) -> None: - assert isinstance(out_stream, int) - data = memoryview(data) - remaining = len(data) - while remaining: - await trio.lowlevel.wait_writable(out_stream) - written = os.write(out_stream, data) - data = data[written:] - remaining -= written - - while True: - data = await in_stream.receive_some() - if not data: - return - await write(data)