Skip to content

Commit

Permalink
Fix linting errors in unit tests of "reinstall()" method
Browse files Browse the repository at this point in the history
  • Loading branch information
Delgan committed Feb 19, 2025
1 parent 2d2b4e0 commit 3cfd03f
Showing 1 changed file with 72 additions and 72 deletions.
144 changes: 72 additions & 72 deletions tests/test_reinstall.py
Original file line number Diff line number Diff line change
@@ -1,72 +1,72 @@
import multiprocessing
import os

import pytest

from loguru import logger


@pytest.fixture
def fork_context():
yield multiprocessing.get_context("fork")


@pytest.fixture
def spawn_context():
yield multiprocessing.get_context("spawn")


class Writer:
def __init__(self):
self._output = ""

def write(self, message):
self._output += message

def read(self):
return self._output


def subworker(logger):
logger.reinstall()
logger.info("Child")
deeper_subworker()


def deeper_subworker():
logger.info("Grandchild")


@pytest.mark.skipif(os.name == "nt", reason="Windows does not support forking")
def test_process_fork(fork_context):
writer = Writer()

logger.add(writer, context=fork_context, format="{message}", enqueue=True, catch=False)

process = fork_context.Process(target=subworker, args=(logger,))
process.start()
process.join()

assert process.exitcode == 0

logger.info("Main")
logger.remove()

assert writer.read() == "Child\nGrandchild\nMain\n"


def test_process_spawn(spawn_context):
writer = Writer()

logger.add(writer, context=spawn_context, format="{message}", enqueue=True, catch=False)

process = spawn_context.Process(target=subworker, args=(logger,))
process.start()
process.join()

assert process.exitcode == 0

logger.info("Main")
logger.remove()

assert writer.read() == "Child\nGrandchild\nMain\n"
import multiprocessing
import os

import pytest

from loguru import logger


@pytest.fixture
def fork_context():
return multiprocessing.get_context("fork")


@pytest.fixture
def spawn_context():
return multiprocessing.get_context("spawn")


class Writer:
def __init__(self):
self._output = ""

def write(self, message):
self._output += message

def read(self):
return self._output


def subworker(logger):
logger.reinstall()
logger.info("Child")
deeper_subworker()


def deeper_subworker():
logger.info("Grandchild")


@pytest.mark.skipif(os.name == "nt", reason="Windows does not support forking")
def test_process_fork(fork_context):
writer = Writer()

logger.add(writer, context=fork_context, format="{message}", enqueue=True, catch=False)

process = fork_context.Process(target=subworker, args=(logger,))
process.start()
process.join()

assert process.exitcode == 0

logger.info("Main")
logger.remove()

assert writer.read() == "Child\nGrandchild\nMain\n"


def test_process_spawn(spawn_context):
writer = Writer()

logger.add(writer, context=spawn_context, format="{message}", enqueue=True, catch=False)

process = spawn_context.Process(target=subworker, args=(logger,))
process.start()
process.join()

assert process.exitcode == 0

logger.info("Main")
logger.remove()

assert writer.read() == "Child\nGrandchild\nMain\n"

0 comments on commit 3cfd03f

Please sign in to comment.