val/irctest
1
0
Fork 0
mirror of https://github.com/progval/irctest.git synced 2025-04-06 07:19:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix lint

Browse Source
This commit is contained in:
Val Lorentz
2024-11-17 18:05:49 +01:00
parent 4983db7cad
commit 895849ed93
2 changed files with 28 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
irctest/basecontrollers.py
View File

@ -11,8 +11,8 @@ import subprocess
import sys import sys
import tempfile import tempfile
import textwrap import textwrap
import time
import threading import threading
import time
from typing import ( from typing import (
IO, IO,
Any, Any,
@ -150,16 +150,21 @@ class _BaseController:
self._own_ports.remove((hostname, port)) self._own_ports.remove((hostname, port))
def execute( def execute(
self, command: Sequence[Union[str, Path]], proc_name: Optional[str], **kwargs: Any self,
command: Sequence[Union[str, Path]],
proc_name: Optional[str] = None,
**kwargs: Any,
) -> subprocess.Popen: ) -> subprocess.Popen:
output_to = None if self.debug_mode else subprocess.DEVNULL output_to = None if self.debug_mode else subprocess.DEVNULL
proc_name = proc_name or command[0] proc_name = proc_name or str(command[0])
kwargs.setdefault("stdout", output_to) kwargs.setdefault("stdout", output_to)
kwargs.setdefault("stderr", output_to) kwargs.setdefault("stderr", output_to)
stream_stdout = stream_stderr = None stream_stdout = stream_stderr = None
if kwargs["stdout"] in (None, subprocess.STDOUT): if kwargs["stdout"] in (None, subprocess.STDOUT):
kwargs["stdout"] = subprocess.PIPE kwargs["stdout"] = subprocess.PIPE
def stream_stdout():
def stream_stdout() -> None:
assert proc.stdout is not None # for mypy
for line in proc.stdout: for line in proc.stdout:
prefix = f"{time.time():.3f} {proc_name} ".encode() prefix = f"{time.time():.3f} {proc_name} ".encode()
try: try:
@ -167,16 +172,20 @@ class _BaseController:
except ValueError: except ValueError:
# "I/O operation on closed file" # "I/O operation on closed file"
pass pass
if kwargs["stderr"] in (subprocess.STDOUT, None): if kwargs["stderr"] in (subprocess.STDOUT, None):
kwargs["stdout"] = subprocess.PIPE kwargs["stdout"] = subprocess.PIPE
def stream_stderr():
for line in proc.stdout: def stream_stderr() -> None:
assert proc.stderr is not None # for mypy
for line in proc.stderr:
prefix = f"{time.time():.3f} {proc_name} ".encode() prefix = f"{time.time():.3f} {proc_name} ".encode()
try: try:
sys.stdout.buffer.write(prefix + line) sys.stdout.buffer.write(prefix + line)
except ValueError: except ValueError:
# "I/O operation on closed file" # "I/O operation on closed file"
pass pass
proc = subprocess.Popen(command, **kwargs) proc = subprocess.Popen(command, **kwargs)
if stream_stdout is not None: if stream_stdout is not None:
threading.Thread(target=stream_stdout, name="stream_stdout").start() threading.Thread(target=stream_stdout, name="stream_stdout").start()

19
irctest/controllers/sable.py
View File

@ -85,7 +85,13 @@ def certs_dir() -> Path:
certs_dir = tempfile.TemporaryDirectory() certs_dir = tempfile.TemporaryDirectory()
(Path(certs_dir.name) / "gen_certs.sh").write_text(GEN_CERTS) (Path(certs_dir.name) / "gen_certs.sh").write_text(GEN_CERTS)
subprocess.run( subprocess.run(
["bash", "gen_certs.sh", "My.Little.Server", "My.Little.History", "My.Little.Services"], [
"bash",
"gen_certs.sh",
"My.Little.Server",
"My.Little.History",
"My.Little.Services",
],
cwd=certs_dir.name, cwd=certs_dir.name,
check=True, check=True,
) )
@ -585,7 +591,9 @@ class SableHistoryController(BaseServicesController):
def run(self, protocol: str, server_hostname: str, server_port: int) -> None: def run(self, protocol: str, server_hostname: str, server_port: int) -> None:
assert protocol == "sable" assert protocol == "sable"
assert self.server_controller.directory is not None assert self.server_controller.directory is not None
history_db_url=os.environ.get("PIFPAF_POSTGRESQL_URL") or os.environ.get("IRCTEST_POSTGRESQL_URL") history_db_url = os.environ.get("PIFPAF_POSTGRESQL_URL") or os.environ.get(
"IRCTEST_POSTGRESQL_URL"
)
assert history_db_url, ( assert history_db_url, (
"Cannot find a postgresql database to use as backend for sable_history. " "Cannot find a postgresql database to use as backend for sable_history. "
"Either set the IRCTEST_POSTGRESQL_URL env var to a libpq URL, or " "Either set the IRCTEST_POSTGRESQL_URL env var to a libpq URL, or "
@ -594,10 +602,9 @@ class SableHistoryController(BaseServicesController):
) )
with self.server_controller.open_file("configs/history_server.conf") as fd: with self.server_controller.open_file("configs/history_server.conf") as fd:
fd.write(HISTORY_SERVER_CONFIG % { vals = dict(self.server_controller.template_vars)
**self.server_controller.template_vars, vals["history_db_url"] = history_db_url
"history_db_url": history_db_url, fd.write(HISTORY_SERVER_CONFIG % vals)
})
self.proc = self.execute( self.proc = self.execute(
[ [