For each service there is a folder called testing with different tests to add to the CI/CD pipe.
During the software development pieces of code used to develop are essential to write short tests. Those routines are put in a separate folder and called prior to commit or during deployment. Ideally the deployment should fail if the test didn’t pass.
Those routine can be unit tests of single functions (especially using classes) or REST request to a service.
import os, sys, gzip, random, csv, json, datetime, re, time
import urllib3, requests
import psycopg2
from sqlalchemy import create_engine
if __name__=="__main__":
= os.environ['DB_NAME'], os.environ['DB_USER'], os.environ['DB_PASS']
db_name, db_user, db_pass = os.environ['DB_HOST'], os.environ['DB_PORT']
db_host, db_port = 'postgresql://{}:{}@{}:{}/{}'.format(db_user, db_pass, db_host, db_port, db_name)
db_string = create_engine(db_string)
db
= db.execute("SELECT datname FROM pg_database;")
res for r in res: print(r)
= db.execute("select * from pg_catalog.pg_tables where schemaname = 'public';")
res for r in res: print(r)
import os, sys, gzip, random, csv, json, datetime, re, time
import urllib3, requests, jwt
def printOutput(resq):
if resq.status_code < 400:
print(resq.json())
else:
print(resq)
print(resq.text)
= "http://0.0.0.0:5005"
url
if __name__=="__main__":
print("test db middleware")
= "http://0.0.0.0:5006"
url = requests.get(url+"/apidocs/")
resq print(resq.status_code)
print("test messaging middleware")
= "http://0.0.0.0:5005/"
url = requests.get(url+"/apidocs/")
resq print(resq.status_code)
print("test push message")
'id'] = inMex['id'] + 1
inMex[= requests.post(url+"/push/"+topic,json=inMex,headers=headers)
resq
printOutput(resq)print("test consume message")
= requests.get(url+"/consume/"+topic,headers=headers,params={"offset":1})
resq
printOutput(resq) print("test consume message")
= requests.get(url+"/latest/"+topic,headers=headers,params={"offset":5})
resq
printOutput(resq)print("test consume message")
= requests.get(url+"/latest_time/"+topic,headers=headers,params={"offset":60*1})
resq
printOutput(resq)print("test list topics")
= requests.get(url+"/topics/",headers=headers)
resq
printOutput(resq)print("test delete")
= requests.get(url+"/delete/"+topic,headers=headers)
resq printOutput(resq)
import os, sys, gzip, random, csv, json, datetime, re, time
import urllib3, requests, jwt
from requests.auth import HTTPBasicAuth
import requests, base64
= "http://0.0.0.0:5006"
url = {'Content-type': 'application/json', 'Accept': 'text/plain'}
headers = {"id":1,"action":"(name, price) values ('test_product',11.25)","table":"products","filter":""}
payload = ""
usrPass
def printOutput(resq):
if resq.status_code < 400:
print(resq.json())
else:
print(resq)
print(resq.reason)
print(resq.text)
if __name__=="__main__":
print("test auth")
= base64.b64encode(usrPass.encode()).decode()
b64Val = requests.get(url+"/authenticate",headers={"Authentication": "%s" % b64Val})
resq print(resq.status_code)
= resq.json()
token "Authentication"] = token
headers[print("test entry")
= requests.get(url+"/book/page/4",headers=headers)
resq
printOutput(resq)= requests.get(url+"/book/page/1",headers=headers)
resq
printOutput(resq)print("test create")
= requests.get(url+"/")
resq print(resq.status_code)
= {"id": 1, "action": "*", "table": "products", "filter": ""}
payload 'action'] = "(name, price) values ('test_product',%f)" % random.uniform(1,30)
payload['filter'] = ""
payload[= requests.post(url+"/create",json=payload,headers=headers)
resq
printOutput(resq)print("test read products")
= {"id": 1, "action": "*", "table": "products", "filter": ""}
payload = requests.post(url+"/select",headers=headers,json=payload)
resq
printOutput(resq)= resq.json()
prodL print("test sql ingestion")
= {"id": 1, "action": "1; drop products;", "table": "products", "filter": ""}
payload = requests.post(url+"/select",headers=headers,json=payload)
resq
printOutput(resq) print("test delete message")
= requests.delete(url+"/delete/"+payload['table']+"/1",headers=headers)
resq for i in prodL:
= requests.delete(url+"/delete/"+payload['table']+"/"+str(i['id']),headers=headers)
resq printOutput(resq)
import json, datetime, base64, sys, os
from time import sleep
import requests, mailparser
from kafka import KafkaConsumer, KafkaProducer, TopicPartition, KafkaAdminClient
= "kafka:9093"
KAFKA_SERVER = "test_topic"
topic = {
inMex "content": "base64-encoded message raw content (unparsed)",
"sender": "sender@example.com",
"recipients": ["recipient1@gmail.com","recipient2@yahoo.com"],
}= "mail-group"
GROUP_ID
def get_producer():
return KafkaProducer(bootstrap_servers=[KAFKA_SERVER],api_version=(0,11,15)
=lambda v: json.dumps(v).encode('utf-8'))
,value_serializer
def get_consumer():
return KafkaConsumer(bootstrap_servers=[KAFKA_SERVER],auto_offset_reset='earliest'
=1000,group_id=GROUP_ID,api_version=(0,11,15)
,consumer_timeout_ms=lambda v: json.loads(v.decode('ascii'))
,value_deserializer#,key_deserializer=lambda v: json.loads(v.decode('ascii'))
=True)#,auto_commit_interval_ms=1000)
,enable_auto_commit
= KafkaAdminClient(bootstrap_servers=[KAFKA_SERVER],api_version=(0,11,15))
admin_client = 'unknown'
topic = 'outbound'
topic
= get_consumer()
consumer = list(consumer.topics())
topicL = TopicPartition(topic, 0)
topic_partition
consumer.assign([topic_partition])
consumer.seek_to_beginning()=topic_partition, offset=0)
consumer.seek(partition= consumer.position(topic_partition)
pos = []
resL for msg in consumer:
resL.append(msg)
print(len(resL))
Docker can test at regular intervals if a service is active
healthcheck:
test: ["CMD", "pg_isready", "-U", "api_ingest"]
interval: 5s
retries: 5