GoldenPogacsa/producer
GoldenPogacsa/producer
Archived
7
0
Fork 0
Code Issues Pull Requests Releases Activity

Create unit tests #3

Merged
tormakris merged 9 commits from testing into master 2020-03-31 21:49:57 +02:00
Conversation 0 Commits 9 Files Changed 8 +12 -24
3 changed files with 12 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit bbfa138849 - Show all commits

2
app.py
View File

@ -23,7 +23,7 @@ sentry_sdk.init("https://3fa5ae886ba1489092ad49a93cb419c1@sentry.kmlabz.com/9")
logging.basicConfig(level=logging.INFO) logging.basicConfig(level=logging.INFO)
LOGGER = logging.getLogger(__name__) LOGGER = logging.getLogger(__name__)
KNOWNCONSUMER= os.environ["PRODUCER_KNOWNCONSUMER"] KNOWNCONSUMER = os.getenv("PRODUCER_KNOWNCONSUMER",'10.69.42.1')
if __name__ == "__main__": if __name__ == "__main__":
LOGGER.info("Producer started") LOGGER.info("Producer started")

2
consumerlocator.py
View File

@ -13,7 +13,7 @@ __copyright__ = "Copyright 2020, GoldenPogácsa Team"
__module_name__ = "consumerlocator" __module_name__ = "consumerlocator"
__version__text__ = "1" __version__text__ = "1"
KNOWNCONSUMER= os.environ["PRODUCER_KNOWNCONSUMER"] KNOWNCONSUMER = os.getenv("PRODUCER_KNOWNCONSUMER",'10.69.42.1')
class ConsumerLocator: class ConsumerLocator:

32
test.py
View File

@ -1,6 +1,5 @@
#!/usr/bin/env python #!/usr/bin/env python
import os
import re import re
import consumerlocator import consumerlocator
import communicator import communicator
@ -16,7 +15,6 @@ __module_name__ = "test"
__version__text__ = "1" __version__text__ = "1"
generateduuid = 'c959ad81-58f9-4445-aab4-8f3d68aee1ad' generateduuid = 'c959ad81-58f9-4445-aab4-8f3d68aee1ad'
os.environ["PRODUCER_KNOWNCONSUMER"] = "10.1.2.3"
def test_generate_string(mocker): def test_generate_string(mocker):
@ -139,16 +137,18 @@ def test_setcurrentconsumer():
def test_learnconsumerlist(httpserver): def test_learnconsumerlist(httpserver):
httpserver.expect_oneshot_request( httpserver.expect_request(
uri="/consumer", uri="/consumer",
method='GET', method='GET',
data="").respond_with_json( data="",
handler_type='permanent').respond_with_json(
["10.69.42.1", "10.10.10.10", "10.20.30.40"]) ["10.69.42.1", "10.10.10.10", "10.20.30.40"])
url = httpserver.url_for("/") url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1) port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
comm = communicator.Communicator( comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}", currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid) uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator( locator = consumerlocator.ConsumerLocator(
uuid=generateduuid, communicator=comm) uuid=generateduuid, communicator=comm)
ret = locator.learnconsumerlist() ret = locator.learnconsumerlist()
@ -162,7 +162,7 @@ def test_getcurrentconsumer(mocker):
uuid=generateduuid) uuid=generateduuid)
locator = consumerlocator.ConsumerLocator( locator = consumerlocator.ConsumerLocator(
uuid=generateduuid, communicator=comm) uuid=generateduuid, communicator=comm)
assert locator.getcurrentconsumer() == "10.1.2.3" assert locator.getcurrentconsumer() == consumerlocator.KNOWNCONSUMER
def test_checkcurrentconsumer(httpserver): def test_checkcurrentconsumer(httpserver):
@ -176,10 +176,11 @@ def test_checkcurrentconsumer(httpserver):
comm = communicator.Communicator( comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}", currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid) uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator( locator = consumerlocator.ConsumerLocator(
uuid=generateduuid, communicator=comm) uuid=generateduuid, communicator=comm)
ret = locator.checkcurrentconsumer() ret = locator.checkcurrentconsumer()
assert ret == False assert ret == True
def test_updateconsumer(httpserver): def test_updateconsumer(httpserver):
@ -193,25 +194,11 @@ def test_updateconsumer(httpserver):
comm = communicator.Communicator( comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}", currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid) uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator( locator = consumerlocator.ConsumerLocator(
uuid=generateduuid, communicator=comm) uuid=generateduuid, communicator=comm)
ret = locator.updateconsumer() ret = locator.updateconsumer()
assert ret == "10.69.42.1" assert ret == f"127.0.0.1:{port}"
httpserver.expect_oneshot_request(
uri="/consumer",
method='GET',
data="").respond_with_json(
[])
url2 = httpserver.url_for("/")
port2 = re.match(r"\W*http[^:]*\D*(\d+)", url2).group(1)
comm2 = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port2}",
uuid=generateduuid)
locator2 = consumerlocator.ConsumerLocator(
uuid=generateduuid, communicator=comm2)
ret2 = locator2.updateconsumer()
assert ret2 is None
def test_updateconsumerlist(httpserver): def test_updateconsumerlist(httpserver):
@ -225,6 +212,7 @@ def test_updateconsumerlist(httpserver):
comm = communicator.Communicator( comm = communicator.Communicator(
currentconsumer=f"127.0.0.1:{port}", currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid) uuid=generateduuid)
consumerlocator.KNOWNCONSUMER = f"127.0.0.1:{port}"
locator = consumerlocator.ConsumerLocator( locator = consumerlocator.ConsumerLocator(
uuid=generateduuid, communicator=comm) uuid=generateduuid, communicator=comm)
ret = locator.updateconsumerlist() ret = locator.updateconsumerlist()