mirror of
https://github.com/miurahr/aqtinstall.git
synced 2025-12-18 05:04:38 +03:00
Monkeypatch multiprocessing.Manager
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
@@ -5,7 +6,7 @@ import textwrap
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from typing import Any, Callable, Dict, Iterable, List, Tuple
|
||||
from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
|
||||
|
||||
import py7zr
|
||||
import pytest
|
||||
@@ -46,6 +47,21 @@ class MockMultiprocessingContext:
|
||||
assert False, "Did not expect to call terminate during unit test"
|
||||
|
||||
|
||||
class MockMultiprocessingManager:
|
||||
class Queue:
|
||||
def __init__(self, *args):
|
||||
pass
|
||||
|
||||
def put_nowait(self, log_record: Optional[logging.LogRecord]):
|
||||
# NOTE: This is certainly not the right way to do this, but it works locally
|
||||
if not log_record or log_record.levelno < logging.INFO:
|
||||
return
|
||||
print(log_record.message, file=sys.stderr)
|
||||
|
||||
def get(self, *args):
|
||||
return None
|
||||
|
||||
|
||||
FILENAME = "filename"
|
||||
UNPATCHED_CONTENT = "unpatched-content"
|
||||
PATCHED_CONTENT = "expected-content"
|
||||
@@ -124,6 +140,11 @@ def apply_mocked_geturl(monkeypatch):
|
||||
lambda *args: MockMultiprocessingContext(),
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
"aqt.installer.multiprocessing.Manager",
|
||||
MockMultiprocessingManager,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"cmd, host, target, version, arch, arch_dir, updates_url, files, expect_out",
|
||||
|
||||
Reference in New Issue
Block a user