epicyon/tests.py

642 lines
25 KiB
Python
Raw Normal View History

2019-06-30 20:14:03 +00:00
__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
2019-07-04 20:36:12 +00:00
import os, os.path
2019-06-30 21:20:02 +00:00
import shutil
2019-06-30 20:14:03 +00:00
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
2019-06-30 21:20:02 +00:00
from daemon import runDaemon
from session import createSession
from posts import deleteAllPosts
from posts import createPublicPost
2019-06-30 22:56:37 +00:00
from posts import sendPost
2019-07-03 10:09:24 +00:00
from posts import archivePosts
2019-07-05 14:39:24 +00:00
from posts import noOfFollowersOnDomain
2019-07-01 11:09:09 +00:00
from follow import clearFollows
from follow import clearFollowers
2019-07-06 19:24:52 +00:00
from utils import followPerson
2019-06-30 21:20:02 +00:00
from follow import followerOfPerson
2019-07-01 11:09:09 +00:00
from follow import unfollowPerson
from follow import unfollowerOfPerson
from follow import getFollowersOfPerson
2019-07-06 13:49:25 +00:00
from follow import sendFollowRequest
2019-07-03 10:04:23 +00:00
from person import createPerson
from person import setPreferredNickname
from person import setBio
2019-07-03 18:24:44 +00:00
from auth import createBasicAuthHeader
from auth import authorizeBasic
from auth import storeBasicCredentials
2019-06-30 20:14:03 +00:00
2019-06-30 21:27:25 +00:00
testServerAliceRunning = False
testServerBobRunning = False
testServerEveRunning = False
2019-06-30 21:27:25 +00:00
2019-06-30 20:14:03 +00:00
def testHttpsigBase(withDigest):
print('testHttpsig(' + str(withDigest) + ')')
2019-07-03 09:40:27 +00:00
nickname='socrates'
2019-06-30 20:14:03 +00:00
domain='argumentative.social'
2019-07-03 19:00:03 +00:00
httpPrefix='https'
2019-07-01 09:31:02 +00:00
port=5576
2019-06-30 22:56:37 +00:00
baseDir=os.getcwd()
2019-07-04 19:09:48 +00:00
password='SuperSecretPassword'
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nickname,domain,port,httpPrefix,False,password)
2019-07-01 09:31:02 +00:00
messageBodyJsonStr = '{"a key": "a value", "another key": "A string"}'
headersDomain=domain
if port!=80 and port !=443:
headersDomain=domain+':'+str(port)
2019-06-30 20:14:03 +00:00
if not withDigest:
2019-07-01 09:31:02 +00:00
headers = {'host': headersDomain}
2019-06-30 20:14:03 +00:00
else:
2019-07-01 09:31:02 +00:00
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
headers = {'host': headersDomain, 'digest': f'SHA-256={bodyDigest}'}
2019-06-30 20:14:03 +00:00
path='/inbox'
2019-07-03 19:00:03 +00:00
signatureHeader = signPostHeaders(privateKeyPem, nickname, domain, port, path, httpPrefix, None)
2019-06-30 20:14:03 +00:00
headers['signature'] = signatureHeader
2019-07-03 19:00:03 +00:00
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, '/inbox' ,False, messageBodyJsonStr)
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, '/parambulator/inbox', False , messageBodyJsonStr) == False
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, '/inbox', True, messageBodyJsonStr) == False
2019-06-30 20:14:03 +00:00
if not withDigest:
# fake domain
headers = {'host': 'bogon.domain'}
else:
# correct domain but fake message
2019-07-01 09:31:02 +00:00
messageBodyJsonStr = '{"a key": "a value", "another key": "Fake GNUs"}'
bodyDigest = base64.b64encode(SHA256.new(messageBodyJsonStr.encode()).digest())
2019-06-30 20:14:03 +00:00
headers = {'host': domain, 'digest': f'SHA-256={bodyDigest}'}
headers['signature'] = signatureHeader
2019-07-03 19:00:03 +00:00
assert verifyPostHeaders(httpPrefix, publicKeyPem, headers, '/inbox', True, messageBodyJsonStr) == False
2019-06-30 20:14:03 +00:00
def testHttpsig():
testHttpsigBase(False)
testHttpsigBase(True)
def testCache():
print('testCache')
personUrl="cat@cardboard.box"
personJson={ "id": 123456, "test": "This is a test" }
2019-07-01 14:30:48 +00:00
personCache={}
storePersonInCache(personUrl,personJson,personCache)
result=getPersonFromCache(personUrl,personCache)
2019-06-30 20:14:03 +00:00
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
2019-06-30 21:20:02 +00:00
def createServerAlice(path: str,domain: str,port: int,federationList: [],ocapGranted: {},hasFollows: bool,hasPosts :bool,ocapAlways: bool):
2019-06-30 21:20:02 +00:00
print('Creating test server: Alice on port '+str(port))
if os.path.isdir(path):
shutil.rmtree(path)
os.mkdir(path)
os.chdir(path)
2019-07-03 09:40:27 +00:00
nickname='alice'
2019-07-04 14:36:29 +00:00
httpPrefix='http'
2019-06-30 21:20:02 +00:00
useTor=False
2019-07-03 15:10:18 +00:00
clientToServer=False
2019-07-04 19:09:48 +00:00
password='alicepass'
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(path,nickname,domain,port,httpPrefix,True,password)
2019-07-04 16:24:23 +00:00
deleteAllPosts(path,nickname,domain,'inbox')
deleteAllPosts(path,nickname,domain,'outbox')
2019-07-06 13:49:25 +00:00
if hasFollows:
2019-07-06 19:24:52 +00:00
followPerson(path,nickname,domain,'bob','127.0.0.100:61936',federationList,True)
followerOfPerson(path,nickname,domain,'bob','127.0.0.100:61936',federationList,True)
2019-07-06 13:49:25 +00:00
if hasPosts:
2019-07-07 11:53:32 +00:00
createPublicPost(path,nickname, domain, port,httpPrefix, "No wise fish would go anywhere without a porpoise", False, True, clientToServer,ocapGranted)
createPublicPost(path,nickname, domain, port,httpPrefix, "Curiouser and curiouser!", False, True, clientToServer,ocapGranted)
createPublicPost(path,nickname, domain, port,httpPrefix, "In the gardens of memory, in the palace of dreams, that is where you and I shall meet", False, True, clientToServer,ocapGranted)
2019-06-30 21:27:25 +00:00
global testServerAliceRunning
testServerAliceRunning = True
2019-06-30 21:20:02 +00:00
print('Server running: Alice')
runDaemon(path,domain,port,httpPrefix,federationList,ocapAlways,ocapGranted,useTor,True)
2019-06-30 21:20:02 +00:00
def createServerBob(path: str,domain: str,port: int,federationList: [],ocapGranted: {},hasFollows: bool,hasPosts :bool,ocapAlways :bool):
2019-06-30 21:20:02 +00:00
print('Creating test server: Bob on port '+str(port))
if os.path.isdir(path):
shutil.rmtree(path)
os.mkdir(path)
os.chdir(path)
2019-07-03 09:40:27 +00:00
nickname='bob'
2019-07-03 19:00:03 +00:00
httpPrefix='http'
2019-06-30 21:20:02 +00:00
useTor=False
2019-07-03 15:10:18 +00:00
clientToServer=False
2019-07-04 19:09:48 +00:00
password='bobpass'
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(path,nickname,domain,port,httpPrefix,True,password)
2019-07-04 16:24:23 +00:00
deleteAllPosts(path,nickname,domain,'inbox')
deleteAllPosts(path,nickname,domain,'outbox')
2019-07-06 13:49:25 +00:00
if hasFollows:
2019-07-06 19:24:52 +00:00
followPerson(path,nickname,domain,'alice','127.0.0.50:61935',federationList,True)
followerOfPerson(path,nickname,domain,'alice','127.0.0.50:61935',federationList,True)
2019-07-06 13:49:25 +00:00
if hasPosts:
2019-07-07 11:53:32 +00:00
createPublicPost(path,nickname, domain, port,httpPrefix, "It's your life, live it your way.", False, True, clientToServer,ocapGranted)
createPublicPost(path,nickname, domain, port,httpPrefix, "One of the things I've realised is that I am very simple", False, True, clientToServer,ocapGranted)
createPublicPost(path,nickname, domain, port,httpPrefix, "Quantum physics is a bit of a passion of mine", False, True, clientToServer,ocapGranted)
2019-06-30 21:27:25 +00:00
global testServerBobRunning
testServerBobRunning = True
2019-06-30 21:20:02 +00:00
print('Server running: Bob')
runDaemon(path,domain,port,httpPrefix,federationList,ocapAlways,ocapGranted,useTor,True)
2019-06-30 21:20:02 +00:00
def createServerEve(path: str,domain: str,port: int,federationList: [],ocapGranted: {},hasFollows: bool,hasPosts :bool,ocapAlways :bool):
print('Creating test server: Eve on port '+str(port))
if os.path.isdir(path):
shutil.rmtree(path)
os.mkdir(path)
os.chdir(path)
nickname='eve'
httpPrefix='http'
useTor=False
clientToServer=False
password='evepass'
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(path,nickname,domain,port,httpPrefix,True,password)
deleteAllPosts(path,nickname,domain,'inbox')
deleteAllPosts(path,nickname,domain,'outbox')
global testServerEveRunning
testServerEveRunning = True
print('Server running: Eve')
runDaemon(path,domain,port,httpPrefix,federationList,ocapAlways,ocapGranted,useTor,True)
2019-06-30 21:20:02 +00:00
def testPostMessageBetweenServers():
print('Testing sending message from one server to the inbox of another')
2019-06-30 21:27:25 +00:00
global testServerAliceRunning
global testServerBobRunning
testServerAliceRunning = False
testServerBobRunning = False
2019-07-03 19:00:03 +00:00
httpPrefix='http'
2019-06-30 22:56:37 +00:00
useTor=False
2019-07-02 09:25:29 +00:00
federationList=['127.0.0.50','127.0.0.100']
2019-07-07 11:53:32 +00:00
ocapGranted={}
2019-06-30 22:56:37 +00:00
2019-06-30 21:20:02 +00:00
baseDir=os.getcwd()
2019-07-06 13:49:25 +00:00
if os.path.isdir(baseDir+'/.tests'):
shutil.rmtree(baseDir+'/.tests')
os.mkdir(baseDir+'/.tests')
2019-06-30 21:20:02 +00:00
ocapAlways=False
2019-06-30 21:20:02 +00:00
# create the servers
2019-06-30 22:56:37 +00:00
aliceDir=baseDir+'/.tests/alice'
2019-07-02 09:25:29 +00:00
aliceDomain='127.0.0.50'
2019-06-30 22:56:37 +00:00
alicePort=61935
thrAlice = threadWithTrace(target=createServerAlice,args=(aliceDir,aliceDomain,alicePort,federationList,ocapGranted,True,True,ocapAlways),daemon=True)
2019-06-30 21:20:02 +00:00
2019-06-30 22:56:37 +00:00
bobDir=baseDir+'/.tests/bob'
2019-07-02 09:25:29 +00:00
bobDomain='127.0.0.100'
2019-06-30 22:56:37 +00:00
bobPort=61936
thrBob = threadWithTrace(target=createServerBob,args=(bobDir,bobDomain,bobPort,federationList,ocapGranted,True,True,ocapAlways),daemon=True)
2019-07-01 21:01:43 +00:00
thrAlice.start()
2019-06-30 21:20:02 +00:00
thrBob.start()
2019-07-01 21:01:43 +00:00
assert thrAlice.isAlive()==True
2019-06-30 21:20:02 +00:00
assert thrBob.isAlive()==True
2019-06-30 21:27:25 +00:00
# wait for both servers to be running
while not (testServerAliceRunning and testServerBobRunning):
time.sleep(1)
2019-07-04 20:25:19 +00:00
time.sleep(1)
2019-06-30 21:20:02 +00:00
2019-06-30 22:56:37 +00:00
print('Alice sends to Bob')
os.chdir(aliceDir)
2019-07-02 09:25:29 +00:00
sessionAlice = createSession(aliceDomain,alicePort,useTor)
2019-06-30 22:56:37 +00:00
inReplyTo=None
inReplyToAtomUri=None
subject=None
aliceSendThreads = []
alicePostLog = []
2019-07-01 12:51:55 +00:00
followersOnly=False
saveToFile=True
2019-07-03 15:10:18 +00:00
clientToServer=False
2019-07-01 12:51:55 +00:00
ccUrl=None
2019-07-01 14:30:48 +00:00
alicePersonCache={}
aliceCachedWebfingers={}
2019-07-07 11:53:32 +00:00
sendResult = sendPost(sessionAlice,aliceDir,'alice', aliceDomain, alicePort, 'bob', bobDomain, bobPort, ccUrl, httpPrefix, 'Why is a mouse when it spins?', followersOnly, saveToFile, clientToServer, federationList, ocapGranted, aliceSendThreads, alicePostLog, aliceCachedWebfingers,alicePersonCache,inReplyTo, inReplyToAtomUri, subject)
2019-06-30 22:56:37 +00:00
print('sendResult: '+str(sendResult))
2019-07-04 20:36:12 +00:00
queuePath=bobDir+'/accounts/bob@'+bobDomain+'/queue'
inboxPath=bobDir+'/accounts/bob@'+bobDomain+'/inbox'
for i in range(10):
if os.path.isdir(inboxPath):
if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])>0:
break
2019-07-01 21:01:43 +00:00
time.sleep(1)
2019-06-30 22:56:37 +00:00
2019-06-30 21:20:02 +00:00
# stop the servers
thrAlice.kill()
thrAlice.join()
assert thrAlice.isAlive()==False
thrBob.kill()
thrBob.join()
assert thrBob.isAlive()==False
2019-07-04 20:36:12 +00:00
# inbox item created
assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1
# queue item removed
assert len([name for name in os.listdir(queuePath) if os.path.isfile(os.path.join(queuePath, name))])==0
os.chdir(baseDir)
shutil.rmtree(aliceDir)
shutil.rmtree(bobDir)
2019-07-06 13:49:25 +00:00
def testFollowBetweenServers():
print('Testing sending a follow request from one server to another')
global testServerAliceRunning
global testServerBobRunning
global testServerEveRunning
2019-07-06 13:49:25 +00:00
testServerAliceRunning = False
testServerBobRunning = False
testServerEveRunning = False
2019-07-06 13:49:25 +00:00
httpPrefix='http'
useTor=False
federationList=[]
2019-07-07 11:53:32 +00:00
ocapGranted={}
2019-07-06 13:49:25 +00:00
baseDir=os.getcwd()
if os.path.isdir(baseDir+'/.tests'):
shutil.rmtree(baseDir+'/.tests')
os.mkdir(baseDir+'/.tests')
ocapAlways=True
2019-07-06 13:49:25 +00:00
# create the servers
aliceDir=baseDir+'/.tests/alice'
aliceDomain='127.0.0.42'
alicePort=61935
thrAlice = threadWithTrace(target=createServerAlice,args=(aliceDir,aliceDomain,alicePort,federationList,ocapGranted,False,False,ocapAlways),daemon=True)
2019-07-06 13:49:25 +00:00
bobDir=baseDir+'/.tests/bob'
bobDomain='127.0.0.64'
bobPort=61936
thrBob = threadWithTrace(target=createServerBob,args=(bobDir,bobDomain,bobPort,federationList,ocapGranted,False,False,ocapAlways),daemon=True)
2019-07-06 13:49:25 +00:00
eveDir=baseDir+'/.tests/eve'
eveDomain='127.0.0.55'
evePort=61937
thrEve = threadWithTrace(target=createServerEve,args=(eveDir,eveDomain,evePort,federationList,ocapGranted,False,False,False),daemon=True)
2019-07-06 13:49:25 +00:00
thrAlice.start()
thrBob.start()
thrEve.start()
2019-07-06 13:49:25 +00:00
assert thrAlice.isAlive()==True
assert thrBob.isAlive()==True
assert thrEve.isAlive()==True
2019-07-06 13:49:25 +00:00
# wait for both servers to be running
ctr=0
while not (testServerAliceRunning and testServerBobRunning and testServerEveRunning):
2019-07-06 13:49:25 +00:00
time.sleep(1)
ctr+=1
if ctr>10:
break
print('Alice online: '+str(testServerAliceRunning))
print('Bob online: '+str(testServerBobRunning))
print('Eve online: '+str(testServerEveRunning))
assert ctr<=10
2019-07-06 13:49:25 +00:00
time.sleep(1)
# In the beginning all was calm and there were no follows
print('Alice sends a follow request to Bob')
print('Both are strictly enforcing object capabilities')
2019-07-06 13:49:25 +00:00
os.chdir(aliceDir)
sessionAlice = createSession(aliceDomain,alicePort,useTor)
inReplyTo=None
inReplyToAtomUri=None
subject=None
aliceSendThreads = []
alicePostLog = []
followersOnly=False
saveToFile=True
clientToServer=False
ccUrl=None
alicePersonCache={}
aliceCachedWebfingers={}
aliceSendThreads=[]
alicePostLog=[]
sendResult = \
sendFollowRequest(sessionAlice,aliceDir, \
'alice',aliceDomain,alicePort,httpPrefix, \
'bob',bobDomain,bobPort,httpPrefix, \
2019-07-07 11:53:32 +00:00
clientToServer,federationList,ocapGranted,
2019-07-06 13:49:25 +00:00
aliceSendThreads,alicePostLog, \
aliceCachedWebfingers,alicePersonCache,True)
print('sendResult: '+str(sendResult))
2019-07-06 17:15:03 +00:00
for t in range(10):
if os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt'):
2019-07-06 19:24:52 +00:00
if os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt'):
if os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/ocap/accept/'+httpPrefix+':##'+aliceDomain+':'+str(alicePort)+'#users#alice.json'):
if os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/ocap/granted/'+httpPrefix+':##'+bobDomain+':'+str(bobPort)+'#users#bob.json'):
2019-07-07 16:06:38 +00:00
break
2019-07-06 17:15:03 +00:00
time.sleep(1)
print('\n\nEve tries to send to Bob')
sessionEve = createSession(eveDomain,evePort,useTor)
eveSendThreads = []
evePostLog = []
evePersonCache={}
eveCachedWebfingers={}
eveSendThreads=[]
evePostLog=[]
sendResult = sendPost(sessionEve,eveDir,'eve', eveDomain, evePort, 'bob', bobDomain, bobPort, ccUrl, httpPrefix, 'Eve message', followersOnly, saveToFile, clientToServer, federationList, ocapGranted, eveSendThreads, evePostLog, eveCachedWebfingers,evePersonCache,inReplyTo, inReplyToAtomUri, subject)
print('sendResult: '+str(sendResult))
queuePath=bobDir+'/accounts/bob@'+bobDomain+'/queue'
inboxPath=bobDir+'/accounts/bob@'+bobDomain+'/inbox'
eveMessageArrived=False
for i in range(5):
time.sleep(1)
if os.path.isdir(inboxPath):
if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])>1:
eveMessageArrived=True
print('Eve message sent to Bob!')
break
# capabilities should have prevented delivery
assert eveMessageArrived==False
print('Message from Eve to Bob was correctly rejected by object capabilities')
aliceSendThreads = []
alicePostLog = []
alicePersonCache={}
aliceCachedWebfingers={}
aliceSendThreads=[]
alicePostLog=[]
sendResult = sendPost(sessionAlice,aliceDir,'alice', aliceDomain, alicePort, 'bob', bobDomain, bobPort, ccUrl, httpPrefix, 'Alice message', followersOnly, saveToFile, clientToServer, federationList, ocapGranted, aliceSendThreads, alicePostLog, aliceCachedWebfingers,alicePersonCache,inReplyTo, inReplyToAtomUri, subject)
print('sendResult: '+str(sendResult))
queuePath=bobDir+'/accounts/bob@'+bobDomain+'/queue'
inboxPath=bobDir+'/accounts/bob@'+bobDomain+'/inbox'
aliceMessageArrived=False
for i in range(5):
time.sleep(1)
if os.path.isdir(inboxPath):
if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])>1:
aliceMessageArrived=True
print('Alice message sent to Bob!')
break
assert aliceMessageArrived==True
print('Message from Alice to Bob succeeded, since it was granted capabilities')
2019-07-06 13:49:25 +00:00
# stop the servers
thrAlice.kill()
thrAlice.join()
assert thrAlice.isAlive()==False
thrBob.kill()
thrBob.join()
assert thrBob.isAlive()==False
assert os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/ocap/accept/'+httpPrefix+':##'+aliceDomain+':'+str(alicePort)+'#users#alice.json')
assert os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/ocap/granted/'+httpPrefix+':##'+bobDomain+':'+str(bobPort)+'#users#bob.json')
2019-07-07 16:33:59 +00:00
2019-07-06 17:15:03 +00:00
assert 'alice@'+aliceDomain in open(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt').read()
assert 'bob@'+bobDomain in open(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt').read()
2019-07-06 13:49:25 +00:00
os.chdir(baseDir)
2019-07-07 16:06:38 +00:00
shutil.rmtree(baseDir+'/.tests')
2019-07-06 13:49:25 +00:00
def testFollowersOfPerson():
print('testFollowersOfPerson')
currDir=os.getcwd()
nickname='mxpop'
domain='diva.domain'
password='birb'
port=80
httpPrefix='https'
federationList=[]
baseDir=currDir+'/.tests_followersofperson'
if os.path.isdir(baseDir):
shutil.rmtree(baseDir)
os.mkdir(baseDir)
os.chdir(baseDir)
createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
createPerson(baseDir,'maxboardroom',domain,port,httpPrefix,True,password)
createPerson(baseDir,'ultrapancake',domain,port,httpPrefix,True,password)
createPerson(baseDir,'drokk',domain,port,httpPrefix,True,password)
createPerson(baseDir,'sausagedog',domain,port,httpPrefix,True,password)
clearFollows(baseDir,nickname,domain)
2019-07-06 19:24:52 +00:00
followPerson(baseDir,nickname,domain,'maxboardroom',domain,federationList,True)
followPerson(baseDir,'drokk',domain,'ultrapancake',domain,federationList,True)
# deliberate duplication
2019-07-06 19:24:52 +00:00
followPerson(baseDir,'drokk',domain,'ultrapancake',domain,federationList,True)
followPerson(baseDir,'sausagedog',domain,'ultrapancake',domain,federationList,True)
followPerson(baseDir,nickname,domain,'ultrapancake',domain,federationList,True)
followPerson(baseDir,nickname,domain,'someother','randodomain.net',federationList,True)
followList=getFollowersOfPerson(baseDir,'ultrapancake',domain)
assert len(followList)==3
assert 'mxpop@'+domain in followList
assert 'drokk@'+domain in followList
assert 'sausagedog@'+domain in followList
os.chdir(currDir)
shutil.rmtree(baseDir)
def testNoOfFollowersOnDomain():
print('testNoOfFollowersOnDomain')
currDir=os.getcwd()
nickname='mxpop'
domain='diva.domain'
otherdomain='soup.dragon'
password='birb'
port=80
httpPrefix='https'
federationList=[]
baseDir=currDir+'/.tests_nooffollowersOndomain'
if os.path.isdir(baseDir):
shutil.rmtree(baseDir)
os.mkdir(baseDir)
os.chdir(baseDir)
createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
createPerson(baseDir,'maxboardroom',otherdomain,port,httpPrefix,True,password)
createPerson(baseDir,'ultrapancake',otherdomain,port,httpPrefix,True,password)
createPerson(baseDir,'drokk',otherdomain,port,httpPrefix,True,password)
createPerson(baseDir,'sausagedog',otherdomain,port,httpPrefix,True,password)
2019-07-06 19:24:52 +00:00
followPerson(baseDir,'drokk',otherdomain,nickname,domain,federationList,True)
followPerson(baseDir,'sausagedog',otherdomain,nickname,domain,federationList,True)
followPerson(baseDir,'maxboardroom',otherdomain,nickname,domain,federationList,True)
2019-07-06 19:24:52 +00:00
followerOfPerson(baseDir,nickname,domain,'cucumber','sandwiches.party',federationList,True)
followerOfPerson(baseDir,nickname,domain,'captainsensible','damned.zone',federationList,True)
followerOfPerson(baseDir,nickname,domain,'pilchard','zombies.attack',federationList,True)
followerOfPerson(baseDir,nickname,domain,'drokk',otherdomain,federationList,True)
followerOfPerson(baseDir,nickname,domain,'sausagedog',otherdomain,federationList,True)
followerOfPerson(baseDir,nickname,domain,'maxboardroom',otherdomain,federationList,True)
followersOnOtherDomain=noOfFollowersOnDomain(baseDir,nickname+'@'+domain, otherdomain)
assert followersOnOtherDomain==3
unfollowerOfPerson(baseDir,nickname,domain,'sausagedog',otherdomain)
followersOnOtherDomain=noOfFollowersOnDomain(baseDir,nickname+'@'+domain, otherdomain)
assert followersOnOtherDomain==2
os.chdir(currDir)
shutil.rmtree(baseDir)
2019-07-03 09:24:55 +00:00
def testFollows():
2019-07-03 10:04:23 +00:00
print('testFollows')
2019-07-03 09:24:55 +00:00
currDir=os.getcwd()
2019-07-03 09:40:27 +00:00
nickname='test529'
2019-07-03 09:24:55 +00:00
domain='testdomain.com'
2019-07-04 19:09:48 +00:00
password='mypass'
2019-07-03 09:24:55 +00:00
port=80
2019-07-03 19:00:03 +00:00
httpPrefix='https'
2019-07-03 09:24:55 +00:00
federationList=['wild.com','mesh.com']
baseDir=currDir+'/.tests_testfollows'
if os.path.isdir(baseDir):
shutil.rmtree(baseDir)
os.mkdir(baseDir)
2019-07-04 19:09:48 +00:00
os.chdir(baseDir)
createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
2019-07-03 09:24:55 +00:00
2019-07-03 09:40:27 +00:00
clearFollows(baseDir,nickname,domain)
2019-07-06 19:24:52 +00:00
followPerson(baseDir,nickname,domain,'badger','wild.com',federationList,False)
followPerson(baseDir,nickname,domain,'squirrel','secret.com',federationList,False)
followPerson(baseDir,nickname,domain,'rodent','drainpipe.com',federationList,False)
followPerson(baseDir,nickname,domain,'batman','mesh.com',federationList,False)
followPerson(baseDir,nickname,domain,'giraffe','trees.com',federationList,False)
2019-07-03 09:24:55 +00:00
2019-07-03 09:40:27 +00:00
f = open(baseDir+'/accounts/'+nickname+'@'+domain+'/following.txt', "r")
2019-07-03 09:33:28 +00:00
domainFound=False
2019-07-03 09:24:55 +00:00
for followingDomain in f:
testDomain=followingDomain.split('@')[1].replace('\n','')
2019-07-03 09:33:28 +00:00
if testDomain=='mesh.com':
domainFound=True
2019-07-03 09:24:55 +00:00
if testDomain not in federationList:
print(testDomain)
assert(False)
2019-07-03 09:33:28 +00:00
assert(domainFound)
2019-07-03 09:40:27 +00:00
unfollowPerson(baseDir,nickname,domain,'batman','mesh.com')
2019-07-03 09:33:28 +00:00
domainFound=False
for followingDomain in f:
testDomain=followingDomain.split('@')[1].replace('\n','')
if testDomain=='mesh.com':
domainFound=True
assert(domainFound==False)
2019-07-03 09:40:27 +00:00
clearFollowers(baseDir,nickname,domain)
2019-07-06 19:24:52 +00:00
followerOfPerson(baseDir,nickname,domain,'badger','wild.com',federationList,False)
followerOfPerson(baseDir,nickname,domain,'squirrel','secret.com',federationList,False)
followerOfPerson(baseDir,nickname,domain,'rodent','drainpipe.com',federationList,False)
followerOfPerson(baseDir,nickname,domain,'batman','mesh.com',federationList,False)
followerOfPerson(baseDir,nickname,domain,'giraffe','trees.com',federationList,False)
2019-07-03 09:24:55 +00:00
2019-07-03 09:40:27 +00:00
f = open(baseDir+'/accounts/'+nickname+'@'+domain+'/followers.txt', "r")
2019-07-03 09:24:55 +00:00
for followerDomain in f:
testDomain=followerDomain.split('@')[1].replace('\n','')
if testDomain not in federationList:
print(testDomain)
assert(False)
os.chdir(currDir)
shutil.rmtree(baseDir)
2019-07-03 10:04:23 +00:00
def testCreatePerson():
print('testCreatePerson')
currDir=os.getcwd()
nickname='test382'
domain='badgerdomain.com'
2019-07-04 19:09:48 +00:00
password='mypass'
2019-07-03 10:04:23 +00:00
port=80
2019-07-03 19:00:03 +00:00
httpPrefix='https'
2019-07-03 15:10:18 +00:00
clientToServer=False
2019-07-03 10:04:23 +00:00
baseDir=currDir+'/.tests_createperson'
if os.path.isdir(baseDir):
shutil.rmtree(baseDir)
os.mkdir(baseDir)
os.chdir(baseDir)
2019-07-03 09:24:55 +00:00
2019-07-04 19:09:48 +00:00
privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
assert os.path.isfile(baseDir+'/accounts/passwords')
2019-07-04 16:24:23 +00:00
deleteAllPosts(baseDir,nickname,domain,'inbox')
deleteAllPosts(baseDir,nickname,domain,'outbox')
2019-07-03 10:04:23 +00:00
setPreferredNickname(baseDir,nickname,domain,'badger')
setBio(baseDir,nickname,domain,'Randomly roaming in your backyard')
2019-07-04 16:24:23 +00:00
archivePosts(nickname,domain,baseDir,'inbox',4)
archivePosts(nickname,domain,baseDir,'outbox',4)
2019-07-03 19:00:03 +00:00
createPublicPost(baseDir,nickname, domain, port,httpPrefix, "G'day world!", False, True, clientToServer, None, None, 'Not suitable for Vogons')
2019-07-03 10:04:23 +00:00
os.chdir(currDir)
shutil.rmtree(baseDir)
2019-07-03 18:24:44 +00:00
def testAuthentication():
print('testAuthentication')
currDir=os.getcwd()
nickname='test8743'
password='SuperSecretPassword12345'
baseDir=currDir+'/.tests_authentication'
if os.path.isdir(baseDir):
shutil.rmtree(baseDir)
os.mkdir(baseDir)
os.chdir(baseDir)
assert storeBasicCredentials(baseDir,'othernick','otherpass')
assert storeBasicCredentials(baseDir,'bad:nick','otherpass')==False
assert storeBasicCredentials(baseDir,'badnick','otherpa:ss')==False
assert storeBasicCredentials(baseDir,nickname,password)
authHeader=createBasicAuthHeader(nickname,password)
2019-07-04 08:56:15 +00:00
assert authorizeBasic(baseDir,'/users/'+nickname+'/inbox',authHeader,False)
assert authorizeBasic(baseDir,'/users/'+nickname,authHeader,False)==False
assert authorizeBasic(baseDir,'/users/othernick/inbox',authHeader,False)==False
2019-07-03 18:24:44 +00:00
authHeader=createBasicAuthHeader(nickname,password+'1')
2019-07-04 08:56:15 +00:00
assert authorizeBasic(baseDir,'/users/'+nickname+'/inbox',authHeader,False)==False
2019-07-03 18:24:44 +00:00
2019-07-03 19:13:23 +00:00
password='someOtherPassword'
assert storeBasicCredentials(baseDir,nickname,password)
authHeader=createBasicAuthHeader(nickname,password)
2019-07-04 08:56:15 +00:00
assert authorizeBasic(baseDir,'/users/'+nickname+'/inbox',authHeader,False)
2019-07-03 19:13:23 +00:00
2019-07-03 18:24:44 +00:00
os.chdir(currDir)
shutil.rmtree(baseDir)
2019-06-30 21:20:02 +00:00
def runAllTests():
print('Running tests...')
testHttpsig()
testCache()
testThreads()
2019-07-03 10:04:23 +00:00
testCreatePerson()
2019-07-03 18:24:44 +00:00
testAuthentication()
testFollowersOfPerson()
testNoOfFollowersOnDomain()
testFollows()
2019-07-03 18:24:44 +00:00
print('Tests succeeded\n')