Skip to content

Commit

Permalink
No need this to be a class
Browse files Browse the repository at this point in the history
  • Loading branch information
hynek committed Aug 2, 2023
1 parent 403f991 commit 78dc242
Showing 1 changed file with 110 additions and 107 deletions.
217 changes: 110 additions & 107 deletions tests/test_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,154 +11,157 @@
from .ifaces import AnotherService, Service, YetAnotherService


class TestIntegration:
def test_register_factory_get(self, registry, container):
"""
register_factory registers a factory and get returns the service.
def test_register_factory_get(registry, container):
"""
register_factory registers a factory and get returns the service.
The service is cached.
"""
registry.register_factory(Service, Service)
The service is cached.
"""
registry.register_factory(Service, Service)

svc = container.get(Service)
svc = container.get(Service)

assert isinstance(svc, Service)
assert svc is container.get(Service)
assert isinstance(svc, Service)
assert svc is container.get(Service)

def test_register_value_get(self, registry, container, svc):
"""
register_value registers a service object and get returns it.
"""
registry.register_value(Service, svc)

assert svc is container.get(Service)
assert svc is container.get(Service)
def test_register_value_get(registry, container, svc):
"""
register_value registers a service object and get returns it.
"""
registry.register_value(Service, svc)

def test_get_not_found(self, container):
"""
Asking for a service that isn't registered raises a ServiceNotFoundError.
"""
with pytest.raises(svcs.exceptions.ServiceNotFoundError) as ei:
container.get(Service)
assert svc is container.get(Service)
assert svc is container.get(Service)

assert Service is ei.value.args[0]

def test_passes_container_bc_name(self, registry, container):
"""
If the factory takes an argument called `svcs_container`, it is passed
on instantiation.
"""
def test_get_not_found(container):
"""
Asking for a service that isn't registered raises a ServiceNotFoundError.
"""
with pytest.raises(svcs.exceptions.ServiceNotFoundError) as ei:
container.get(Service)

def factory(svcs_container):
return str(svcs_container.get(int))
assert Service is ei.value.args[0]

registry.register_value(int, 42)
registry.register_factory(str, factory)

assert "42" == container.get(str)
def test_passes_container_bc_name(registry, container):
"""
If the factory takes an argument called `svcs_container`, it is passed on
instantiation.
"""

def test_passes_container_bc_annotation(self, registry, container):
"""
If the factory takes an argument annotated with svcs.Container, it is
passed on instantiation.
"""
def factory(svcs_container):
return str(svcs_container.get(int))

def factory(foo: svcs.Container):
return str(foo.get(int))
registry.register_value(int, 42)
registry.register_factory(str, factory)

registry.register_value(int, 42)
registry.register_factory(str, factory)
assert "42" == container.get(str)

assert "42" == container.get(str)

def test_get_pings(self, registry, container, svc):
"""
get_pings returns a list of ServicePings.
"""
registry.register_factory(AnotherService, AnotherService)
registry.register_value(Service, svc, ping=lambda _: None)
def test_passes_container_bc_annotation(registry, container):
"""
If the factory takes an argument annotated with svcs.Container, it is
passed on instantiation.
"""

assert [Service] == [
ping._rs.svc_type for ping in container.get_pings()
]
def factory(foo: svcs.Container):
return str(foo.get(int))

def test_cleanup_called(self, registry, container):
"""
Services that have a cleanup have them called on cleanup.
"""
cleaned_up = False
registry.register_value(int, 42)
registry.register_factory(str, factory)

def factory():
nonlocal cleaned_up
yield 42
cleaned_up = True
assert "42" == container.get(str)

registry.register_factory(Service, factory)

container.get(Service)
def test_get_pings(registry, container, svc):
"""
get_pings returns a list of ServicePings.
"""
registry.register_factory(AnotherService, AnotherService)
registry.register_value(Service, svc, ping=lambda _: None)

assert not cleaned_up
assert [Service] == [ping._rs.svc_type for ping in container.get_pings()]

container.close()

assert cleaned_up
assert not container._instantiated
assert not container._on_close
def test_cleanup_called(registry, container):
"""
Services that have a cleanup have them called on cleanup.
"""
cleaned_up = False

def test_close_resilient(self, container, registry, caplog):
"""
Failing cleanups are logged and ignored. They do not break the
cleanup process.
"""
def factory():
nonlocal cleaned_up
yield 42
cleaned_up = True

def factory():
yield 1
raise Exception
registry.register_factory(Service, factory)

cleaned_up = False
container.get(Service)

def factory_no_boom():
nonlocal cleaned_up
assert not cleaned_up

yield 3
container.close()

cleaned_up = True
assert cleaned_up
assert not container._instantiated
assert not container._on_close

registry.register_factory(Service, factory)
registry.register_factory(YetAnotherService, factory_no_boom)

assert 1 == container.get(Service)
assert 3 == container.get(YetAnotherService)
def test_close_resilient(container, registry, caplog):
"""
Failing cleanups are logged and ignored. They do not break the cleanup
process.
"""

assert not cleaned_up
def factory():
yield 1
raise Exception

container.close()
cleaned_up = False

assert "tests.ifaces.Service" == caplog.records[0].svcs_service_name
assert cleaned_up
def factory_no_boom():
nonlocal cleaned_up

def test_warns_if_generator_does_not_stop_after_cleanup(
self, registry, container
):
"""
If a generator doesn't stop after cleanup, a warning is emitted.
"""
yield 3

def factory():
yield Service()
yield 42
cleaned_up = True

registry.register_factory(Service, factory)
registry.register_factory(Service, factory)
registry.register_factory(YetAnotherService, factory_no_boom)

container.get(Service)
assert 1 == container.get(Service)
assert 3 == container.get(YetAnotherService)

with pytest.warns(UserWarning) as wi:
container.close()
assert not cleaned_up

assert (
"Container clean up for 'tests.ifaces.Service' "
"didn't stop iterating." == wi.pop().message.args[0]
)
container.close()

assert "tests.ifaces.Service" == caplog.records[0].svcs_service_name
assert cleaned_up


def test_warns_if_generator_does_not_stop_after_cleanup(registry, container):
"""
If a generator doesn't stop after cleanup, a warning is emitted.
"""

def factory():
yield Service()
yield 42

registry.register_factory(Service, factory)

container.get(Service)

with pytest.warns(UserWarning) as wi:
container.close()

assert (
"Container clean up for 'tests.ifaces.Service' "
"didn't stop iterating." == wi.pop().message.args[0]
)


@pytest.mark.asyncio()
Expand Down

0 comments on commit 78dc242

Please sign in to comment.