From 8705b9e44c6bab1ed9cf2f0ad0c8470d4827aa64 Mon Sep 17 00:00:00 2001 From: Bob Mottram Date: Sun, 30 Jun 2019 21:14:03 +0100 Subject: [PATCH] Move tests to their own file --- cache.py | 9 ------- daemon.py | 6 ++--- epicyon.py | 2 +- httpsig.py | 34 ------------------------- tests.py | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ threads.py | 14 ----------- 6 files changed, 78 insertions(+), 61 deletions(-) create mode 100644 tests.py diff --git a/cache.py b/cache.py index 01f875f45..6ea13ea3e 100644 --- a/cache.py +++ b/cache.py @@ -46,13 +46,4 @@ def getWebfingerFromCache(handle: str): if cachedWebfingers.get(handle): return cachedWebfingers[handle] return None - -def testCache(): - print('testCache') - personUrl="cat@cardboard.box" - personJson={ "id": 123456, "test": "This is a test" } - storePersonInCache(personUrl,personJson) - result=getPersonFromCache(personUrl) - assert result['id']==123456 - assert result['test']=='This is a test' diff --git a/daemon.py b/daemon.py index 5dc71fb4f..bfa796f47 100644 --- a/daemon.py +++ b/daemon.py @@ -12,8 +12,9 @@ import json import cgi from pprint import pprint from session import createSession -from httpsig import testHttpsig -from cache import testCache +from tests import testHttpsig +from tests import testCache +from tests import testThreads from webfinger import webfingerMeta from webfinger import webfingerLookup from person import personLookup @@ -21,7 +22,6 @@ from person import personKeyLookup from person import personOutboxJson from inbox import inboxPermittedMessage from follow import getFollowingFeed -from threads import testThreads import os import sys diff --git a/epicyon.py b/epicyon.py index 47039885b..e739575d6 100644 --- a/epicyon.py +++ b/epicyon.py @@ -22,7 +22,7 @@ import json import sys import requests from pprint import pprint -from httpsig import testHttpsig +from tests import testHttpsig from daemon import runDaemon import socket from follow import clearFollows diff --git a/httpsig.py b/httpsig.py index 4ae9ad083..6e7564173 100644 --- a/httpsig.py +++ b/httpsig.py @@ -108,37 +108,3 @@ def verifyPostHeaders(https: bool, publicKeyPem: str, headers: dict, path: str, return True except (ValueError, TypeError): return False - -def testHttpsigBase(withDigest): - print('testHttpsig(' + str(withDigest) + ')') - username='socrates' - domain='argumentative.social' - https=True - port=80 - privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(username,domain,port,https,False) - messageBodyJson = '{"a key": "a value", "another key": "A string"}' - if not withDigest: - headers = {'host': domain} - else: - bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest()) - headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'} - path='/inbox' - signatureHeader = signPostHeaders(privateKeyPem, username, domain, path, https, None) - headers['signature'] = signatureHeader - assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox' ,False, messageBodyJson) - assert verifyPostHeaders(https, publicKeyPem, headers, '/parambulator/inbox', False , messageBodyJson) == False - assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJson) == False - if not withDigest: - # fake domain - headers = {'host': 'bogon.domain'} - else: - # correct domain but fake message - messageBodyJson = '{"a key": "a value", "another key": "Fake GNUs"}' - bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest()) - headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'} - headers['signature'] = signatureHeader - assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJson) == False - -def testHttpsig(): - testHttpsigBase(False) - testHttpsigBase(True) diff --git a/tests.py b/tests.py new file mode 100644 index 000000000..0b22ee680 --- /dev/null +++ b/tests.py @@ -0,0 +1,74 @@ +__filename__ = "tests.py" +__author__ = "Bob Mottram" +__license__ = "AGPL3+" +__version__ = "0.0.1" +__maintainer__ = "Bob Mottram" +__email__ = "bob@freedombone.net" +__status__ = "Production" + +import base64 +import time +from person import createPerson +from Crypto.Hash import SHA256 +from httpsig import signPostHeaders +from httpsig import verifyPostHeaders +from cache import storePersonInCache +from cache import getPersonFromCache +from threads import threadWithTrace + +def testHttpsigBase(withDigest): + print('testHttpsig(' + str(withDigest) + ')') + username='socrates' + domain='argumentative.social' + https=True + port=80 + privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(username,domain,port,https,False) + messageBodyJson = '{"a key": "a value", "another key": "A string"}' + if not withDigest: + headers = {'host': domain} + else: + bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest()) + headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'} + path='/inbox' + signatureHeader = signPostHeaders(privateKeyPem, username, domain, path, https, None) + headers['signature'] = signatureHeader + assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox' ,False, messageBodyJson) + assert verifyPostHeaders(https, publicKeyPem, headers, '/parambulator/inbox', False , messageBodyJson) == False + assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJson) == False + if not withDigest: + # fake domain + headers = {'host': 'bogon.domain'} + else: + # correct domain but fake message + messageBodyJson = '{"a key": "a value", "another key": "Fake GNUs"}' + bodyDigest = base64.b64encode(SHA256.new(messageBodyJson.encode()).digest()) + headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'} + headers['signature'] = signatureHeader + assert verifyPostHeaders(https, publicKeyPem, headers, '/inbox', True, messageBodyJson) == False + +def testHttpsig(): + testHttpsigBase(False) + testHttpsigBase(True) + +def testCache(): + print('testCache') + personUrl="cat@cardboard.box" + personJson={ "id": 123456, "test": "This is a test" } + storePersonInCache(personUrl,personJson) + result=getPersonFromCache(personUrl) + assert result['id']==123456 + assert result['test']=='This is a test' + +def testThreadsFunction(param: str): + for i in range(10000): + time.sleep(2) + +def testThreads(): + print('testThreads') + thr = threadWithTrace(target=testThreadsFunction,args=('test',),daemon=True) + thr.start() + assert thr.isAlive()==True + time.sleep(1) + thr.kill() + thr.join() + assert thr.isAlive()==False diff --git a/threads.py b/threads.py index 4a6323b76..e639da134 100644 --- a/threads.py +++ b/threads.py @@ -40,17 +40,3 @@ class threadWithTrace(threading.Thread): def kill(self): self.killed = True - -def testThreadsFunction(param: str): - for i in range(10000): - time.sleep(2) - -def testThreads(): - print('testThreads') - thr = threadWithTrace(target=testThreadsFunction,args=('test',),daemon=True) - thr.start() - assert thr.isAlive()==True - time.sleep(1) - thr.kill() - thr.join() - assert thr.isAlive()==False