Migrate misc wiring tests

This commit is contained in:
Roman Mogylatov 2021-10-13 19:11:45 -04:00
parent a7450605b5
commit 7c8cbdcfd9
6 changed files with 100 additions and 171 deletions

View File

@ -3,7 +3,7 @@
from decimal import Decimal
from dependency_injector import errors
from dependency_injector.wiring import Provide, Provider, wire
from dependency_injector.wiring import Closing, Provide, Provider, wire
from pytest import fixture, mark, raises
from wiringsamples import module, package, resourceclosing
@ -289,6 +289,23 @@ def test_closing_resource():
assert result_1 is not result_2
@mark.usefixtures("resourceclosing_container")
def test_closing_resource_bypass_marker_injection():
resourceclosing.Service.reset_counter()
result_1 = resourceclosing.test_function(service=Closing[Provide[resourceclosing.Container.service]])
assert isinstance(result_1, resourceclosing.Service)
assert result_1.init_counter == 1
assert result_1.shutdown_counter == 1
result_2 = resourceclosing.test_function(service=Closing[Provide[resourceclosing.Container.service]])
assert isinstance(result_2, resourceclosing.Service)
assert result_2.init_counter == 2
assert result_2.shutdown_counter == 2
assert result_1 is not result_2
@mark.usefixtures("resourceclosing_container")
def test_closing_resource_context():
resourceclosing.Service.reset_counter()
@ -313,3 +330,8 @@ def test_class_decorator():
def test_container():
service = module.test_container()
assert isinstance(service, Service)
def test_bypass_marker_injection():
service = module.test_function(service=Provide[Container.service])
assert isinstance(service, Service)

View File

@ -0,0 +1,32 @@
"""Tests for wiring with dynamic container."""
from dependency_injector import containers, providers
from pytest import fixture
from wiringstringidssamples import module, package
from wiringstringidssamples.service import Service
@fixture(autouse=True)
def container():
sub = containers.DynamicContainer()
sub.int_object = providers.Object(1)
container = containers.DynamicContainer()
container.config = providers.Configuration()
container.service = providers.Factory(Service)
container.sub = sub
container.wire(
modules=[module],
packages=[package],
)
yield container
container.unwire()
def test_wire():
service = module.test_function()
assert isinstance(service, Service)

View File

@ -3,7 +3,7 @@
from decimal import Decimal
from dependency_injector import errors
from dependency_injector.wiring import Provide, Provider, wire
from dependency_injector.wiring import Closing, Provide, Provider, wire
from pytest import fixture, mark, raises
from wiringstringidssamples import module, package, resourceclosing
@ -289,6 +289,23 @@ def test_closing_resource():
assert result_1 is not result_2
@mark.usefixtures("resourceclosing_container")
def test_closing_resource_bypass_marker_injection():
resourceclosing.Service.reset_counter()
result_1 = resourceclosing.test_function(service=Closing[Provide["service"]])
assert isinstance(result_1, resourceclosing.Service)
assert result_1.init_counter == 1
assert result_1.shutdown_counter == 1
result_2 = resourceclosing.test_function(service=Closing[Provide["service"]])
assert isinstance(result_2, resourceclosing.Service)
assert result_2.init_counter == 2
assert result_2.shutdown_counter == 2
assert result_1 is not result_2
@mark.usefixtures("resourceclosing_container")
def test_closing_resource_context():
resourceclosing.Service.reset_counter()
@ -313,3 +330,8 @@ def test_class_decorator():
def test_container():
service = module.test_container()
assert isinstance(service, Service)
def test_bypass_marker_injection():
service = module.test_function(service=Provide["service"])
assert isinstance(service, Service)

View File

@ -1,81 +0,0 @@
import unittest
from dependency_injector.wiring import (
Provide,
Closing,
)
# Runtime import to avoid syntax errors in samples on Python < 3.5
import os
_TOP_DIR = os.path.abspath(
os.path.sep.join((
os.path.dirname(__file__),
"../",
)),
)
_SAMPLES_DIR = os.path.abspath(
os.path.sep.join((
os.path.dirname(__file__),
"../samples/",
)),
)
import sys
sys.path.append(_TOP_DIR)
sys.path.append(_SAMPLES_DIR)
from wiringsamples import module
from wiringsamples.service import Service
from wiringsamples.container import Container
class WiringAndQueue(unittest.TestCase):
def test_wire_queue(self) -> None:
from wiringsamples import queuemodule
container = Container()
self.addCleanup(container.unwire)
# Should not raise exception
# See: https://github.com/ets-labs/python-dependency-injector/issues/362
try:
container.wire(modules=[queuemodule])
except:
raise
class WiringAndFastAPITest(unittest.TestCase):
container: Container
def test_bypass_marker_injection(self):
container = Container()
container.wire(modules=[module])
self.addCleanup(container.unwire)
service = module.test_function(service=Provide[Container.service])
self.assertIsInstance(service, Service)
def test_closing_resource_bypass_marker_injection(self):
from wiringsamples import resourceclosing
resourceclosing.Service.reset_counter()
container = resourceclosing.Container()
container.wire(modules=[resourceclosing])
self.addCleanup(container.unwire)
result_1 = resourceclosing.test_function(
service=Closing[Provide[resourceclosing.Container.service]],
)
self.assertIsInstance(result_1, resourceclosing.Service)
self.assertEqual(result_1.init_counter, 1)
self.assertEqual(result_1.shutdown_counter, 1)
result_2 = resourceclosing.test_function(
service=Closing[Provide[resourceclosing.Container.service]],
)
self.assertIsInstance(result_2, resourceclosing.Service)
self.assertEqual(result_2.init_counter, 2)
self.assertEqual(result_2.shutdown_counter, 2)
self.assertIsNot(result_1, result_2)

View File

@ -1,88 +0,0 @@
import unittest
from dependency_injector.wiring import (
Provide,
Closing,
)
from dependency_injector import containers, providers, errors
# Runtime import to avoid syntax errors in samples on Python < 3.5
import os
_TOP_DIR = os.path.abspath(
os.path.sep.join((
os.path.dirname(__file__),
"../",
)),
)
_SAMPLES_DIR = os.path.abspath(
os.path.sep.join((
os.path.dirname(__file__),
"../samples/",
)),
)
import sys
sys.path.append(_TOP_DIR)
sys.path.append(_SAMPLES_DIR)
from wiringstringidssamples import module, package
from wiringstringidssamples.service import Service
from wiringstringidssamples.container import Container
class WiringAndFastAPITest(unittest.TestCase):
container: Container
def test_bypass_marker_injection(self):
container = Container()
container.wire(modules=[module])
self.addCleanup(container.unwire)
service = module.test_function(service=Provide[Container.service])
self.assertIsInstance(service, Service)
def test_closing_resource_bypass_marker_injection(self):
from wiringstringidssamples import resourceclosing
resourceclosing.Service.reset_counter()
container = resourceclosing.Container()
container.wire(modules=[resourceclosing])
self.addCleanup(container.unwire)
result_1 = resourceclosing.test_function(
service=Closing[Provide[resourceclosing.Container.service]],
)
self.assertIsInstance(result_1, resourceclosing.Service)
self.assertEqual(result_1.init_counter, 1)
self.assertEqual(result_1.shutdown_counter, 1)
result_2 = resourceclosing.test_function(
service=Closing[Provide[resourceclosing.Container.service]],
)
self.assertIsInstance(result_2, resourceclosing.Service)
self.assertEqual(result_2.init_counter, 2)
self.assertEqual(result_2.shutdown_counter, 2)
self.assertIsNot(result_1, result_2)
class WireDynamicContainerTest(unittest.TestCase):
def test_wire(self):
sub = containers.DynamicContainer()
sub.int_object = providers.Object(1)
container = containers.DynamicContainer()
container.config = providers.Configuration()
container.service = providers.Factory(Service)
container.sub = sub
container.wire(
modules=[module],
packages=[package],
)
self.addCleanup(container.unwire)
service = module.test_function()
self.assertIsInstance(service, Service)

View File

@ -0,0 +1,22 @@
"""Tests for wiring causes no issues with queue.Queue from std lib."""
from pytest import fixture
from wiringsamples import queuemodule
from wiringsamples.container import Container
@fixture
def container():
container = Container()
yield container
container.unwire()
def test_wire_queue(container: Container):
# See: https://github.com/ets-labs/python-dependency-injector/issues/362
# Should not raise exception
try:
container.wire(modules=[queuemodule])
except:
raise