mirror of
https://github.com/frappe/frappe_docker.git
synced 2026-06-17 21:55:09 +00:00
99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
import os
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from contextlib import suppress
|
|
from http.client import HTTPResponse
|
|
from typing import Callable, Optional
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
CI = os.getenv("CI")
|
|
|
|
|
|
class Compose:
|
|
def __init__(self, project_name: str, env_file: str):
|
|
self.project_name = project_name
|
|
self.base_cmd = (
|
|
"docker",
|
|
"compose",
|
|
"-p",
|
|
project_name,
|
|
"--env-file",
|
|
env_file,
|
|
)
|
|
|
|
def __call__(self, *cmd: str) -> None:
|
|
file_args = [
|
|
"-f",
|
|
"compose.yaml",
|
|
"-f",
|
|
"overrides/compose.proxy.yaml",
|
|
"-f",
|
|
"overrides/compose.mariadb.yaml",
|
|
"-f",
|
|
"overrides/compose.redis.yaml",
|
|
]
|
|
if CI:
|
|
file_args += ("-f", "tests/compose.ci.yaml")
|
|
|
|
args = self.base_cmd + tuple(file_args) + cmd
|
|
subprocess.check_call(args)
|
|
|
|
def exec(self, *cmd: str) -> None:
|
|
if sys.stdout.isatty():
|
|
self("exec", *cmd)
|
|
else:
|
|
self("exec", "-T", *cmd)
|
|
|
|
def stop(self) -> None:
|
|
# Stop all containers in `test` project if they are running.
|
|
# We don't care if it fails.
|
|
with suppress(subprocess.CalledProcessError):
|
|
subprocess.check_call(self.base_cmd + ("down", "-v", "--remove-orphans"))
|
|
|
|
def bench(self, *cmd: str) -> None:
|
|
self.exec("backend", "bench", *cmd)
|
|
|
|
|
|
def check_url_content(
|
|
url: str, callback: Callable[[str], Optional[str]], site_name: str
|
|
):
|
|
for _ in range(100):
|
|
try:
|
|
response = wait_for_url(url=url, site_name=site_name, attempts=1)
|
|
except RuntimeError:
|
|
pass
|
|
else:
|
|
text: str = response.read().decode()
|
|
ret = callback(text)
|
|
if ret:
|
|
print(ret)
|
|
return
|
|
|
|
time.sleep(0.1)
|
|
|
|
raise RuntimeError(f"Couldn't verify expected content from {url}")
|
|
|
|
|
|
def wait_for_url(url: str, site_name: str, attempts: int = 100) -> HTTPResponse:
|
|
request = Request(url, headers={"Host": site_name})
|
|
|
|
# This is needed to check https override
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
for _ in range(attempts):
|
|
try:
|
|
return urlopen(request, context=ctx)
|
|
except HTTPError as exc:
|
|
if exc.code not in (404, 502):
|
|
raise
|
|
except URLError:
|
|
pass
|
|
|
|
time.sleep(0.1)
|
|
|
|
raise RuntimeError(f"Couldn't ping {url}")
|