Created integration test

This commit is contained in:
Scharnitzky Donát 2020-05-08 20:57:23 +02:00
parent 4610c7a42f
commit f015eb2301
2 changed files with 63 additions and 6 deletions

View File

@ -21,6 +21,14 @@ steps:
- pip3 install -r requirements.txt
- pytest test.py
- name: integration_test
image: python:3.8
environment:
PRODUCER_REDIS: cache
commands:
- pip3 install -r requirements.txt
- pytest integtest.py
- name: build-app
image: banzaicloud/drone-kaniko
settings:

View File

@ -5,6 +5,8 @@ Integration test for the producer module.
"""
import os
import re
import datetime
from communicator import Communicator
from consumerlocator import ConsumerLocator
from messagesender import MessageSender
@ -20,15 +22,62 @@ generateduuid = '2fbff1f2-27e7-44c8-88d9-7348cee8c1c3'
redis_proc = factories.redis_proc(host='cache', port=6379)
redis_db = factories.redisdb('redis_nooproc')
KNOWNCONSUMER = os.getenv("PRODUCER_KNOWNCONSUMER", '10.69.42.1')
answer = ""
def answermethod(Request):
global answer
answer = Request
return ""
def test_integration(mocker):
def test_integration(httpserver):
"""
Tests the whole system.
:param mocker: patches the :class:`communicator.Communicator`.
:param httpserver: simple HTTP server
"""
communicator = Communicator(currentconsumer=KNOWNCONSUMER, uuid=generateduuid)
messagesender = MessageSender(communicator=communicator, uuid=generateduuid)
consumerlocator = ConsumerLocator(uuid=generateduuid, communicator=communicator,
#Inint
httpserver.expect_request(
uri="/consumers",
method='GET',
data="").respond_with_json(
["localhost", "localhost", "1.2.3.4"])
httpserver.expect_request(
uri="/log",
method='POST').respond_with_handler(answermethod)
url = httpserver.url_for("/")
port = re.match(r"\W*http[^:]*\D*(\d+)", url).group(1)
comm = Communicator(
currentconsumer=f"127.0.0.1:{port}",
uuid=generateduuid)
messagesender = MessageSender(communicator=comm, uuid=generateduuid)
consumerlocator = ConsumerLocator(uuid=generateduuid, communicator=comm,
redisconnector=RedisConnector())
#First test method
consumerlocator.learnconsumerlist()
conslist = consumerlocator.red.consumerlist
assert len(conslist) == 3
#Mock in port numbers
consrepllist = []
for consumer in conslist:
newconsumer = {"Host": consumer["Host"]+":"+port, "State": consumer["State"], "LastOk": consumer["LastOk"]}
consrepllist.append(newconsumer)
consumerlocator.red.consumerlist = consrepllist
conslist = consumerlocator.red.consumerlist
#Second test method call
consumerlocator.updateconsumerlist()
conslist = consumerlocator.red.consumerlist
error = False
for consumer in conslist:
if consumer["Host"] == "localhost:"+port and consumer["State"] is True:
pass
elif consumer["Host"] == "1.2.3.4:"+port and consumer["State"] is False:
pass
elif consumer["Host"] == "10.69.42.1:"+port and consumer["State"] is False:
pass
else:
error = True
assert error is False
#Third test method call
consumerlocator.updateconsumer()
assert consumerlocator.red.currentconsumer["Host"] == "localhost:"+port
messagesender.sendmessage()
assert answer != ""