<s> from oslo_utils import reflection <EOL> from taskflow . engines . worker_based import types as worker_types <EOL> from taskflow import test <EOL> from taskflow . test import mock <EOL> from taskflow . tests import utils <EOL> class TestTopicWorker ( test . TestCase ) : <EOL> def test_topic_worker ( self ) : <EOL> worker = worker_types . TopicWorker ( "<STR_LIT>" , <EOL> [ utils . DummyTask ] , identity = "<STR_LIT>" ) <EOL> self . assertTrue ( worker . performs ( utils . DummyTask ) ) <EOL> self . assertFalse ( worker . performs ( utils . NastyTask ) ) <EOL> self . assertEqual ( '<STR_LIT>' , worker . identity ) <EOL> self . assertEqual ( '<STR_LIT>' , worker . topic ) <EOL> class TestProxyFinder ( test . TestCase ) : <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_expiry ( self , mock_now ) : <EOL> finder = worker_types . ProxyWorkerFinder ( '<STR_LIT>' , mock . MagicMock ( ) , [ ] , <EOL> worker_expiry = <NUM_LIT> ) <EOL> w , emit = finder . _add ( '<STR_LIT>' , [ utils . DummyTask ] ) <EOL> w . last_seen = <NUM_LIT:0> <EOL> mock_now . side_effect = [ <NUM_LIT> ] <EOL> gone = finder . clean ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , finder . total_workers ) <EOL> self . assertEqual ( <NUM_LIT:1> , gone ) <EOL> def test_single_topic_worker ( self ) : <EOL> finder = worker_types . ProxyWorkerFinder ( '<STR_LIT>' , mock . MagicMock ( ) , [ ] ) <EOL> w , emit = finder . _add ( '<STR_LIT>' , [ utils . DummyTask ] ) <EOL> self . assertIsNotNone ( w ) <EOL> self . assertTrue ( emit ) <EOL> self . assertEqual ( <NUM_LIT:1> , finder . total_workers ) <EOL> w2 = finder . get_worker_for_task ( utils . DummyTask ) <EOL> self . assertEqual ( w . identity , w2 . identity ) <EOL> def test_multi_same_topic_workers ( self ) : <EOL> finder = worker_types . ProxyWorkerFinder ( '<STR_LIT>' , mock . MagicMock ( ) , [ ] ) <EOL> w , emit = finder . _add ( '<STR_LIT>' , [ utils . DummyTask ] ) <EOL> self . assertIsNotNone ( w ) <EOL> self . assertTrue ( emit ) <EOL> w2 , emit = finder . _add ( '<STR_LIT>' , [ utils . DummyTask ] ) <EOL> self . assertIsNotNone ( w2 ) <EOL> self . assertTrue ( emit ) <EOL> w3 = finder . get_worker_for_task ( <EOL> reflection . get_class_name ( utils . DummyTask ) ) <EOL> self . assertIn ( w3 . identity , [ w . identity , w2 . identity ] ) <EOL> def test_multi_different_topic_workers ( self ) : <EOL> finder = worker_types . ProxyWorkerFinder ( '<STR_LIT>' , mock . MagicMock ( ) , [ ] ) <EOL> added = [ ] <EOL> added . append ( finder . _add ( '<STR_LIT>' , [ utils . DummyTask ] ) ) <EOL> added . append ( finder . _add ( '<STR_LIT>' , [ utils . DummyTask ] ) ) <EOL> added . append ( finder . _add ( '<STR_LIT>' , [ utils . NastyTask ] ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , finder . total_workers ) <EOL> w = finder . get_worker_for_task ( utils . NastyTask ) <EOL> self . assertEqual ( added [ - <NUM_LIT:1> ] [ <NUM_LIT:0> ] . identity , w . identity ) <EOL> w = finder . get_worker_for_task ( utils . DummyTask ) <EOL> self . assertIn ( w . identity , [ w_a [ <NUM_LIT:0> ] . identity for w_a in added [ <NUM_LIT:0> : <NUM_LIT:2> ] ] ) </s>
