KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use subscriber method in localstack

Best Python code snippet using localstack_python

network_flow_random.py

Source:network_flow_random.py

1import random2class NetworkFlowRandom:3    def __init__(self):4        pass5    def apply(self, network_configuration, firedex_configuration, experiment_configuration):6        network_flows = range(10001, 10001 + firedex_configuration["network_flows"])7        result = []8        experiment_subscribers = experiment_configuration["subscribers"]9        for experiment_subscriber in experiment_subscribers:10            subscriber = experiment_subscriber["subscriber"]11            identifier = subscriber["identifier"]12            host = subscriber["host"]13            for port in network_flows:14                subscriber_network_flow = {15                    "identifier": identifier,16                    "network_flow": {17                        "host": host,18                        "port": port,19                        "utility_function": 020                    },21                    "subscriptions": []22                }23                result.append(subscriber_network_flow)24        for experiment_subscriber in experiment_subscribers:25            subscriber = experiment_subscriber["subscriber"]26            identifier = subscriber["identifier"]27            host = subscriber["host"]28            subscriber_network_flows = self.__subscriber_network_flows(result, identifier)29            experiment_subscriptions = subscriber["subscriptions"]30            for experiment_subscription in experiment_subscriptions:31                topic = experiment_subscription["topic"]32                utility_function = experiment_subscription["utilityFunction"]33                port = random.sample(network_flows, 1)[0]34                subscriber_network_flow = self.__subscriber_network_flow(subscriber_network_flows, host, port)35                network_flow = subscriber_network_flow["network_flow"]36                subscriptions = subscriber_network_flow["subscriptions"]37                network_flow["utility_function"] = network_flow["utility_function"] + utility_function38                subscriptions.append({39                    "topic": topic,40                    "utility_function": utility_function41                })42        result = self.__remove_empty_network_flows(result)43        return result44    def __subscriber_network_flows(self, result, identifier):45        subscriber_network_flows = []46        for subscriber_network_flow in result:47            subscriber_identifer = subscriber_network_flow["identifier"]48            if subscriber_identifer == identifier:49                subscriber_network_flows.append(subscriber_network_flow)50        return subscriber_network_flows51    def __subscriber_network_flow(self, subscriber_network_flows, host, port):52        for subscriber_network_flow in subscriber_network_flows:53            subscriber_host = subscriber_network_flow["network_flow"]["host"]54            subscriber_port = subscriber_network_flow["network_flow"]["port"]55            if subscriber_host == host and subscriber_port == port:56                return subscriber_network_flow57        return None58    def __remove_empty_network_flows(self, result):59        new_result = []60        for subscriber_network_flow in result:61            subscriptions = subscriber_network_flow["subscriptions"]62            if subscriptions:63                new_result.append(subscriber_network_flow)...

mailing-list.py

Source:mailing-list.py

2from typing import List3class MailingListPublisherAbc(ABC):4    """Publisher interface with a set of methods for managing subscribers"""5    @abstractmethod6    def add_subscriber(self, subscriber):7        raise NotImplementedError8    @abstractmethod9    def remove_subscriber(self, subscriber):10        raise NotImplementedError11    @abstractmethod12    def notify(self):13        raise NotImplementedError14    @abstractmethod15    def process_latest_content(self) -> str:16        raise NotImplementedError17class SubscriberAbc(ABC):18    """Subscriber interface to interact with publisher"""19    @abstractmethod20    def interact_with_new_content(self, publisher: MailingListPublisherAbc):21        raise NotImplementedError22class MailingListPublisher(MailingListPublisherAbc):23    """Publisher of mailing list"""24    _latest_content: str = ''25    _subscribers: List[SubscriberAbc] = []26    def add_subscriber(self, subscriber: SubscriberAbc) -> None:27        self._subscribers.append(subscriber)28    def remove_subscriber(self, subscriber: SubscriberAbc) -> None:29        try:30            self._subscribers.remove(subscriber)31        except ValueError:32            print('Tried to remove nonexisting subscriber!')33    def update_latest_content(self, new_content: str):34        self._latest_content = new_content35        self.notify()36    def notify(self) -> None:37        for s in self._subscribers:38            s.interact_with_new_content(self)39    def process_latest_content(self):40        return f'This is the latest content ð®: {self._latest_content}'41class SubscriberReaderKind(SubscriberAbc):42    """This is the kind of subscriber which always reads the latest content"""43    def interact_with_new_content(self, publisher: MailingListPublisherAbc):44        content = publisher.process_latest_content()45        self.read(content)46    @staticmethod47    def read(content: str):48        print(content)49        print('Reader says: humm, how interesting...')50class SubscriberProcrastinatorKind(SubscriberAbc):51    """This guy never reads, always saying: latter... ð"""52    def interact_with_new_content(self, publisher: MailingListPublisherAbc):53        content = publisher.process_latest_content()54        print('Procrastinator says: Nah! I\'ll read latter. ðª')55def main():56    publisher = MailingListPublisher()57    reader = SubscriberReaderKind()58    procrastinator = SubscriberProcrastinatorKind()59    publisher.add_subscriber(reader)60    publisher.add_subscriber(procrastinator)61    publisher.update_latest_content('Programmers are filthy rich! ð¤')62    publisher.remove_subscriber(procrastinator)63    print('Bye bye procrastinator! Well, this guy never reads anyway! ð¤·ââï¸')64if __name__ == '__main__':...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.