Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_loads_subscriptions_when_they_are_class_based(self):
subscriptions = load_subscriptions_from_paths(
["tests.subs"],
sub_prefix="test",
filter_by=[lambda attrs: attrs.get("lang") == "en"],
)
assert len(subscriptions) == 2
klass_sub = subscriptions[0]
assert isinstance(klass_sub, Subscription)
assert klass_sub.name == "test-alternative-cool-topic"
assert klass_sub({"id": 4}, lang="en") == 4
def subscriptions(self):
return load_subscriptions_from_paths(
["tests.test_config"],
sub_prefix="test",
filter_by=[lambda args: args.get("lang") == "en"],
)
def handle(self, *args, **options):
headers = ["Topic", "Subscriber(s)", "Sub"]
subscription_paths = discover_subs_modules()
subs = sorted(
load_subscriptions_from_paths(subscription_paths),
key=lambda sub: sub.topic,
)
sub_data = [(sub.topic, sub.name, sub._func.__name__) for sub in subs]
self.stdout.write(tabulate(sub_data, headers=headers))