Spaces:
Runtime error
Runtime error
File size: 4,127 Bytes
4a51346 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
from chromadb.config import Component, System, Settings
from overrides import overrides
from threading import local
import random
data = local() # use thread local just in case tests ever run in parallel
def reset() -> None:
global data
data.starts = []
data.stops = []
data.inits = []
class ComponentA(Component):
def __init__(self, system: System):
data.inits += "A"
super().__init__(system)
self.require(ComponentB)
self.require(ComponentC)
@overrides
def start(self) -> None:
data.starts += "A"
@overrides
def stop(self) -> None:
data.stops += "A"
class ComponentB(Component):
def __init__(self, system: System):
data.inits += "B"
super().__init__(system)
self.require(ComponentC)
self.require(ComponentD)
@overrides
def start(self) -> None:
data.starts += "B"
@overrides
def stop(self) -> None:
data.stops += "B"
class ComponentC(Component):
def __init__(self, system: System):
data.inits += "C"
super().__init__(system)
self.require(ComponentD)
@overrides
def start(self) -> None:
data.starts += "C"
@overrides
def stop(self) -> None:
data.stops += "C"
class ComponentD(Component):
def __init__(self, system: System):
data.inits += "D"
super().__init__(system)
@overrides
def start(self) -> None:
data.starts += "D"
@overrides
def stop(self) -> None:
data.stops += "D"
# Dependency Graph for tests:
# ┌───┐
# │ A │
# └┬─┬┘
# │┌▽──┐
# ││ B │
# │└┬─┬┘
# ┌▽─▽┐│
# │ C ││
# └┬──┘│
# ┌▽───▽┐
# │ D │
# └─────┘
def test_leaf_only() -> None:
settings = Settings()
system = System(settings)
reset()
d = system.instance(ComponentD)
assert isinstance(d, ComponentD)
assert data.inits == ["D"]
system.start()
assert data.starts == ["D"]
system.stop()
assert data.stops == ["D"]
def test_partial() -> None:
settings = Settings()
system = System(settings)
reset()
c = system.instance(ComponentC)
assert isinstance(c, ComponentC)
assert data.inits == ["C", "D"]
system.start()
assert data.starts == ["D", "C"]
system.stop()
assert data.stops == ["C", "D"]
def test_system_startup() -> None:
settings = Settings()
system = System(settings)
reset()
a = system.instance(ComponentA)
assert isinstance(a, ComponentA)
assert data.inits == ["A", "B", "C", "D"]
system.start()
assert data.starts == ["D", "C", "B", "A"]
system.stop()
assert data.stops == ["A", "B", "C", "D"]
def test_system_override_order() -> None:
settings = Settings()
system = System(settings)
reset()
system.instance(ComponentA)
# Deterministically shuffle the instances map to prove that topsort is actually
# working and not just implicitly working because of insertion order.
# This causes the test to actually fail if the deps are not wired up correctly.
random.seed(0)
entries = list(system._instances.items())
random.shuffle(entries)
system._instances = {k: v for k, v in entries}
system.start()
assert data.starts == ["D", "C", "B", "A"]
system.stop()
assert data.stops == ["A", "B", "C", "D"]
class ComponentZ(Component):
def __init__(self, system: System):
super().__init__(system)
self.require(ComponentC)
@overrides
def start(self) -> None:
pass
@overrides
def stop(self) -> None:
pass
def test_runtime_dependencies() -> None:
settings = Settings()
system = System(settings)
reset()
# Nothing to do, no components were requested prior to start
system.start()
assert data.starts == []
# Constructs dependencies and starts them in the correct order
ComponentZ(system)
assert data.starts == ["D", "C"]
system.stop()
assert data.stops == ["C", "D"]
|