test_containers.py

import gc
import time
import weakref

import objgraph
from dependency_injector import containers, providers

from railbird import util
from railbird.containers import RBDeps, RequestContainer


class Parent(containers.DeclarativeContainer):
    __self__ = providers.Self()
    config = providers.Configuration(default={"a": "b"})

    a_factory = providers.Factory(lambda container: container.config.a(), __self__)
    a_singleton = providers.Singleton(lambda container: container.config.a(), __self__)


class Child(containers.DeclarativeContainer):
    a_parent = providers.Provider()

    p = util.ProvidedProxy.from_provider(a_parent)

    a_thing = providers.Singleton(lambda value: value, p.a_singleton)


def test_request_containers_are_garbage_collected():
    testing_container = RBDeps()
    testing_container.init_resources()
    count = 500
    for _ in range(count):
        RequestContainer(rb_deps=testing_container)

    time.sleep(1)
    gc.collect()
    assert len(objgraph.by_type("DynamicContainer")) < 50


def test_parent_container_allocation(testing_container):
    parent = Parent()
    count = 70
    for _ in range(count):
        print(Child(a_parent=parent).a_thing())
    time.sleep(1)
    gc.collect()
    assert len(objgraph.by_type("DynamicContainer")) < 50


def test_child_container_allocation(testing_container):
    parent = Parent()
    for i in range(2):
        Child(a_parent=parent)
    time.sleep(1)
    gc.collect()

    child_container = weakref.ref(
        next(c for c in objgraph.by_type("DynamicContainer") if hasattr(c, "a_parent"))
    )
    time.sleep(1)
    gc.collect()
    print(
        objgraph.show_backrefs(
            child_container(), max_depth=1000000, too_many=100000000, refcounts=True
        )
    )
    chain = objgraph.find_backref_chain(
        child_container(),
        objgraph.is_proper_module,
        max_depth=10000,
        too_many=100000000,
    )
    import ipdb

    ipdb.set_trace()
    assert len(objgraph.by_type("DynamicContainer")) < 50