123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392 |
- __filename__ = "tests.py"
- __author__ = "Bob Mottram"
- __license__ = "AGPL3+"
- __version__ = "1.0.0"
- __maintainer__ = "Bob Mottram"
- __email__ = "bob@freedombone.net"
- __status__ = "Production"
- import base64
- import time
- import os, os.path
- import shutil
- import commentjson
- import json
- from time import gmtime, strftime
- from pprint import pprint
- from person import createPerson
- from Crypto.Hash import SHA256
- from httpsig import signPostHeaders
- from httpsig import verifyPostHeaders
- from httpsig import messageContentDigest
- from cache import storePersonInCache
- from cache import getPersonFromCache
- from threads import threadWithTrace
- from daemon import runDaemon
- from session import createSession
- from posts import deleteAllPosts
- from posts import createPublicPost
- from posts import sendPost
- from posts import archivePosts
- from posts import noOfFollowersOnDomain
- from posts import groupFollowersByDomain
- from posts import sendCapabilitiesUpdate
- from posts import archivePostsForPerson
- from posts import sendPostViaServer
- from follow import clearFollows
- from follow import clearFollowers
- from follow import sendFollowRequestViaServer
- from follow import sendUnfollowRequestViaServer
- from utils import followPerson
- from utils import getNicknameFromActor
- from utils import getDomainFromActor
- from follow import followerOfPerson
- from follow import unfollowPerson
- from follow import unfollowerOfPerson
- from follow import getFollowersOfPerson
- from follow import sendFollowRequest
- from person import createPerson
- from person import setDisplayNickname
- from person import setBio
- from skills import setSkillLevel
- from roles import setRole
- from roles import getRoles
- from roles import outboxDelegate
- from auth import createBasicAuthHeader
- from auth import authorizeBasic
- from auth import storeBasicCredentials
- from like import likePost
- from like import sendLikeViaServer
- from announce import announcePublic
- from announce import sendAnnounceViaServer
- from media import getMediaPath
- from media import getAttachmentMediaType
- from delete import sendDeleteViaServer
- from inbox import validInbox
- from inbox import validInboxFilenames
- from content import addWebLinks
- testServerAliceRunning = False
- testServerBobRunning = False
- testServerEveRunning = False
- def testHttpsigBase(withDigest):
- print('testHttpsig(' + str(withDigest) + ')')
- baseDir=os.getcwd()
- path=baseDir+'/.testHttpsigBase'
- if os.path.isdir(path):
- shutil.rmtree(path)
- os.mkdir(path)
- os.chdir(path)
- contentType='application/activity+json'
- nickname='socrates'
- domain='argumentative.social'
- httpPrefix='https'
- port=5576
- password='SuperSecretPassword'
- privateKeyPem,publicKeyPem,person,wfEndpoint= \
- createPerson(path,nickname,domain,port,httpPrefix,False,password)
- assert privateKeyPem
- messageBodyJson = {"a key": "a value", "another key": "A string","yet another key": "Another string"}
- messageBodyJsonStr=json.dumps(messageBodyJson)
- headersDomain=domain
- if port:
- if port!=80 and port !=443:
- if ':' not in domain:
- headersDomain=domain+':'+str(port)
- dateStr=strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime())
- boxpath='/inbox'
- if not withDigest:
- headers = {'host': headersDomain,'date': dateStr,'content-type': 'application/json'}
- signatureHeader = \
- signPostHeaders(dateStr,privateKeyPem, nickname, \
- domain, port, \
- domain, port, \
- boxpath, httpPrefix, None)
- else:
- bodyDigest = messageContentDigest(messageBodyJsonStr)
- headers = {'host': headersDomain,'date': dateStr,'digest': f'SHA-256={bodyDigest}','content-type': contentType}
- signatureHeader = \
- signPostHeaders(dateStr,privateKeyPem, nickname, \
- domain, port, \
- domain, port, \
- boxpath, httpPrefix, messageBodyJsonStr)
- headers['signature'] = signatureHeader
- assert verifyPostHeaders(httpPrefix,publicKeyPem,headers, \
- boxpath,False,None, \
- messageBodyJsonStr)
- assert verifyPostHeaders(httpPrefix,publicKeyPem,headers, \
- '/parambulator'+boxpath,False,None, \
- messageBodyJsonStr) == False
- assert verifyPostHeaders(httpPrefix,publicKeyPem,headers, \
- boxpath,True,None, \
- messageBodyJsonStr) == False
- if not withDigest:
- # fake domain
- headers = {'host': 'bogon.domain','date': dateStr,'content-type': 'application/json'}
- else:
- # correct domain but fake message
- messageBodyJsonStr = '{"a key": "a value", "another key": "Fake GNUs", "yet another key": "More Fake GNUs"}'
- bodyDigest = messageContentDigest(messageBodyJsonStr)
- headers = {'host': domain,'date': dateStr,'digest': f'SHA-256={bodyDigest}','content-type': contentType}
- headers['signature'] = signatureHeader
- assert verifyPostHeaders(httpPrefix,publicKeyPem,headers, \
- boxpath,True,None, \
- messageBodyJsonStr) == False
- os.chdir(baseDir)
- shutil.rmtree(path)
- def testHttpsig():
- testHttpsigBase(True)
- testHttpsigBase(False)
- def testCache():
- print('testCache')
- personUrl="cat@cardboard.box"
- personJson={ "id": 123456, "test": "This is a test" }
- personCache={}
- storePersonInCache(None,personUrl,personJson,personCache)
- result=getPersonFromCache(None,personUrl,personCache)
- 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
- def createServerAlice(path: str,domain: str,port: int,federationList: [], \
- hasFollows: bool,hasPosts :bool,ocapAlways: bool):
- print('Creating test server: Alice on port '+str(port))
- if os.path.isdir(path):
- shutil.rmtree(path)
- os.mkdir(path)
- os.chdir(path)
- nickname='alice'
- httpPrefix='http'
- useTor=False
- password='alicepass'
- noreply=False
- nolike=False
- nopics=False
- noannounce=False
- cw=False
- useBlurhash=True
- maxReplies=64
- domainMaxPostsPerDay=1000
- accountMaxPostsPerDay=1000
- allowDeletion=True
- privateKeyPem,publicKeyPem,person,wfEndpoint= \
- createPerson(path,nickname,domain,port,httpPrefix,True,password)
- deleteAllPosts(path,nickname,domain,'inbox')
- deleteAllPosts(path,nickname,domain,'outbox')
- assert setSkillLevel(path,nickname,domain,'hacking',90)
- assert setRole(path,nickname,domain,'someproject','guru')
- if hasFollows:
- followPerson(path,nickname,domain,'bob','127.0.0.100:61936', \
- federationList,False)
- followerOfPerson(path,nickname,domain,'bob','127.0.0.100:61936', \
- federationList,False)
- if hasPosts:
- createPublicPost(path,nickname, domain, port,httpPrefix, \
- "No wise fish would go anywhere without a porpoise", \
- False, True, clientToServer,None,None,useBlurhash)
- createPublicPost(path,nickname, domain, port,httpPrefix, \
- "Curiouser and curiouser!", False, True, \
- clientToServer,None,None,useBlurhash)
- 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,None,None,useBlurhash)
- global testServerAliceRunning
- testServerAliceRunning = True
- print('Server running: Alice')
- runDaemon(__version__,"instanceId",False,path,domain,port,port, \
- httpPrefix,federationList, \
- noreply,nolike,nopics,noannounce,cw,ocapAlways, \
- useTor,maxReplies, \
- domainMaxPostsPerDay,accountMaxPostsPerDay, \
- allowDeletion,True,True,False)
- def createServerBob(path: str,domain: str,port: int,federationList: [], \
- hasFollows: bool,hasPosts :bool,ocapAlways :bool):
- print('Creating test server: Bob on port '+str(port))
- if os.path.isdir(path):
- shutil.rmtree(path)
- os.mkdir(path)
- os.chdir(path)
- nickname='bob'
- httpPrefix='http'
- useTor=False
- clientToServer=False
- password='bobpass'
- noreply=False
- nolike=False
- nopics=False
- noannounce=False
- cw=False
- useBlurhash=False
- maxReplies=64
- domainMaxPostsPerDay=1000
- accountMaxPostsPerDay=1000
- allowDeletion=True
- privateKeyPem,publicKeyPem,person,wfEndpoint= \
- createPerson(path,nickname,domain,port,httpPrefix,True,password)
- deleteAllPosts(path,nickname,domain,'inbox')
- deleteAllPosts(path,nickname,domain,'outbox')
- assert setRole(path,nickname,domain,'bandname','bass player')
- assert setRole(path,nickname,domain,'bandname','publicist')
- if hasFollows:
- followPerson(path,nickname,domain, \
- 'alice','127.0.0.50:61935',federationList,False)
- followerOfPerson(path,nickname,domain, \
- 'alice','127.0.0.50:61935',federationList,False)
- if hasPosts:
- createPublicPost(path,nickname, domain, port,httpPrefix, \
- "It's your life, live it your way.", \
- False, True, clientToServer,None,None,useBlurhash)
- createPublicPost(path,nickname, domain, port,httpPrefix, \
- "One of the things I've realised is that I am very simple", \
- False, True, clientToServer,None,None,useBlurhash)
- createPublicPost(path,nickname, domain, port,httpPrefix, \
- "Quantum physics is a bit of a passion of mine", \
- False, True, clientToServer,None,None,useBlurhash)
- global testServerBobRunning
- testServerBobRunning = True
- print('Server running: Bob')
- runDaemon(__version__,"instanceId",False,path,domain,port,port, \
- httpPrefix,federationList, \
- noreply,nolike,nopics,noannounce,cw,ocapAlways, \
- useTor,maxReplies, \
- domainMaxPostsPerDay,accountMaxPostsPerDay, \
- allowDeletion,True,True,False)
- def createServerEve(path: str,domain: str,port: int,federationList: [], \
- 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'
- noreply=False
- nolike=False
- nopics=False
- noannounce=False
- cw=False
- maxReplies=64
- allowDeletion=True
- 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(__version__,"instanceId",False,path,domain,port,port, \
- httpPrefix,federationList, \
- noreply,nolike,nopics,noannounce,cw,ocapAlways, \
- useTor,maxReplies,allowDeletion,True,True,False)
- def testPostMessageBetweenServers():
- print('Testing sending message from one server to the inbox of another')
- global testServerAliceRunning
- global testServerBobRunning
- testServerAliceRunning = False
- testServerBobRunning = False
- httpPrefix='http'
- useTor=False
- baseDir=os.getcwd()
- if os.path.isdir(baseDir+'/.tests'):
- shutil.rmtree(baseDir+'/.tests')
- os.mkdir(baseDir+'/.tests')
- ocapAlways=False
-
- # create the servers
- aliceDir=baseDir+'/.tests/alice'
- aliceDomain='127.0.0.50'
- alicePort=61935
- bobDir=baseDir+'/.tests/bob'
- bobDomain='127.0.0.100'
- bobPort=61936
- federationList=[bobDomain,aliceDomain]
- thrAlice = \
- threadWithTrace(target=createServerAlice, \
- args=(aliceDir,aliceDomain,alicePort, \
- federationList,False,False, \
- ocapAlways),daemon=True)
- thrBob = \
- threadWithTrace(target=createServerBob, \
- args=(bobDir,bobDomain,bobPort, \
- federationList,False,False, \
- ocapAlways),daemon=True)
- thrAlice.start()
- thrBob.start()
- assert thrAlice.isAlive()==True
- assert thrBob.isAlive()==True
- # wait for both servers to be running
- while not (testServerAliceRunning and testServerBobRunning):
- time.sleep(1)
-
- time.sleep(1)
- print('\n\n*******************************************************')
- print('Alice sends to Bob')
- 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={}
- attachedImageFilename=baseDir+'/img/logo.png'
- mediaType=getAttachmentMediaType(attachedImageFilename)
- attachedImageDescription='Logo'
- useBlurhash=True
- # nothing in Alice's outbox
- outboxPath=aliceDir+'/accounts/alice@'+aliceDomain+'/outbox'
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==0
- sendResult = \
- sendPost(__version__, \
- sessionAlice,aliceDir,'alice', aliceDomain, alicePort, \
- 'bob', bobDomain, bobPort, ccUrl, httpPrefix, \
- 'Why is a mouse when it spins? #sillyquestion', followersOnly, \
- saveToFile, clientToServer,attachedImageFilename,mediaType, \
- attachedImageDescription,useBlurhash, federationList, \
- aliceSendThreads, alicePostLog, aliceCachedWebfingers, \
- alicePersonCache,inReplyTo, inReplyToAtomUri, subject)
- print('sendResult: '+str(sendResult))
- queuePath=bobDir+'/accounts/bob@'+bobDomain+'/queue'
- inboxPath=bobDir+'/accounts/bob@'+bobDomain+'/inbox'
- mPath=getMediaPath()
- mediaPath=aliceDir+'/'+mPath
- for i in range(30):
- if os.path.isdir(inboxPath):
- if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])>0:
- if len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==1:
- if len([name for name in os.listdir(mediaPath) if os.path.isfile(os.path.join(mediaPath, name))])>0:
- break
- time.sleep(1)
- # Image attachment created
- assert len([name for name in os.listdir(mediaPath) if os.path.isfile(os.path.join(mediaPath, name))])>0
- # 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
- assert validInbox(bobDir,'bob',bobDomain)
- assert validInboxFilenames(bobDir,'bob',bobDomain,aliceDomain,alicePort)
- print('\n\n*******************************************************')
- print("Bob likes Alice's post")
- followerOfPerson(bobDir,'bob',bobDomain,'alice', \
- aliceDomain+':'+str(alicePort),federationList,False)
- followPerson(aliceDir,'alice',aliceDomain,'bob', \
- bobDomain+':'+str(bobPort),federationList,False)
- sessionBob = createSession(bobDomain,bobPort,useTor)
- bobSendThreads = []
- bobPostLog = []
- bobPersonCache={}
- bobCachedWebfingers={}
- statusNumber=None
- outboxPostFilename=None
- outboxPath=aliceDir+'/accounts/alice@'+aliceDomain+'/outbox'
- for name in os.listdir(outboxPath):
- if '#statuses#' in name:
- statusNumber=int(name.split('#statuses#')[1].replace('.json',''))
- outboxPostFilename=outboxPath+'/'+name
- assert statusNumber>0
- assert outboxPostFilename
- assert likePost(sessionBob,bobDir,federationList, \
- 'bob',bobDomain,bobPort,httpPrefix, \
- 'alice',aliceDomain,alicePort,[], \
- statusNumber,False,bobSendThreads,bobPostLog, \
- bobPersonCache,bobCachedWebfingers, \
- True,__version__)
- for i in range(20):
- if 'likes' in open(outboxPostFilename).read():
- break
- time.sleep(1)
- with open(outboxPostFilename, 'r') as fp:
- alicePostJson=commentjson.load(fp)
- pprint(alicePostJson)
- assert 'likes' in open(outboxPostFilename).read()
- print('\n\n*******************************************************')
- print("Bob repeats Alice's post")
- objectUrl=httpPrefix+'://'+aliceDomain+':'+str(alicePort)+'/users/alice/statuses/'+str(statusNumber)
- inboxPath=aliceDir+'/accounts/alice@'+aliceDomain+'/inbox'
- outboxPath=bobDir+'/accounts/bob@'+bobDomain+'/outbox'
- outboxBeforeAnnounceCount=len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])
- beforeAnnounceCount=len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])
- assert beforeAnnounceCount==0
- print('inbox items before announce: '+str(beforeAnnounceCount))
- announcePublic(sessionBob,bobDir,federationList, \
- 'bob',bobDomain,bobPort,httpPrefix, \
- objectUrl, \
- False,bobSendThreads,bobPostLog, \
- bobPersonCache,bobCachedWebfingers, \
- True,__version__)
- announceMessageArrived=False
- for i in range(10):
- 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))])>0:
- announceMessageArrived=True
- print('Announce message sent to Alice!')
- break
- afterAnnounceCount=len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])
- outboxAfterAnnounceCount=len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])
- print('inbox items after announce: '+str(afterAnnounceCount))
- assert afterAnnounceCount==beforeAnnounceCount+1
- assert outboxAfterAnnounceCount==outboxBeforeAnnounceCount+1
- # stop the servers
- thrAlice.kill()
- thrAlice.join()
- assert thrAlice.isAlive()==False
- thrBob.kill()
- thrBob.join()
- assert thrBob.isAlive()==False
- os.chdir(baseDir)
- shutil.rmtree(aliceDir)
- shutil.rmtree(bobDir)
- def testFollowBetweenServers():
- print('Testing sending a follow request from one server to another')
- global testServerAliceRunning
- global testServerBobRunning
- global testServerEveRunning
- testServerAliceRunning = False
- testServerBobRunning = False
- testServerEveRunning = False
- httpPrefix='http'
- useTor=False
- federationList=[]
- baseDir=os.getcwd()
- if os.path.isdir(baseDir+'/.tests'):
- shutil.rmtree(baseDir+'/.tests')
- os.mkdir(baseDir+'/.tests')
- ocapAlways=True
- # create the servers
- aliceDir=baseDir+'/.tests/alice'
- aliceDomain='127.0.0.42'
- alicePort=61935
- thrAlice = \
- threadWithTrace(target=createServerAlice, \
- args=(aliceDir,aliceDomain,alicePort, \
- federationList,False,False, \
- ocapAlways),daemon=True)
- bobDir=baseDir+'/.tests/bob'
- bobDomain='127.0.0.64'
- bobPort=61936
- thrBob = \
- threadWithTrace(target=createServerBob, \
- args=(bobDir,bobDomain,bobPort, \
- federationList,False,False, \
- ocapAlways),daemon=True)
- eveDir=baseDir+'/.tests/eve'
- eveDomain='127.0.0.55'
- evePort=61937
- thrEve = \
- threadWithTrace(target=createServerEve, \
- args=(eveDir,eveDomain,evePort, \
- federationList,False,False, \
- False),daemon=True)
- thrAlice.start()
- thrBob.start()
- thrEve.start()
- assert thrAlice.isAlive()==True
- assert thrBob.isAlive()==True
- assert thrEve.isAlive()==True
- # wait for all servers to be running
- ctr=0
- while not (testServerAliceRunning and testServerBobRunning and testServerEveRunning):
- time.sleep(1)
- ctr+=1
- if ctr>60:
- break
- print('Alice online: '+str(testServerAliceRunning))
- print('Bob online: '+str(testServerBobRunning))
- print('Eve online: '+str(testServerEveRunning))
- assert ctr<=60
- time.sleep(1)
- # In the beginning all was calm and there were no follows
- print('*********************************************************')
- print('Alice sends a follow request to Bob')
- print('Both are strictly enforcing object capabilities')
- 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, \
- clientToServer,federationList, \
- aliceSendThreads,alicePostLog, \
- aliceCachedWebfingers,alicePersonCache, \
- True,__version__)
- print('sendResult: '+str(sendResult))
- bobCapsFilename=bobDir+'/accounts/bob@'+bobDomain+'/ocap/accept/'+httpPrefix+':##'+aliceDomain+':'+str(alicePort)+'#users#alice.json'
- aliceCapsFilename=aliceDir+'/accounts/alice@'+aliceDomain+'/ocap/granted/'+httpPrefix+':##'+bobDomain+':'+str(bobPort)+'#users#bob.json'
- for t in range(10):
- if os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt'):
- if os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt'):
- if os.path.isfile(bobCapsFilename):
- if os.path.isfile(aliceCapsFilename):
- break
- time.sleep(1)
- with open(bobCapsFilename, 'r') as fp:
- bobCapsJson=commentjson.load(fp)
- if not bobCapsJson.get('capability'):
- print("Unexpected format for Bob's capabilities")
- pprint(bobCapsJson)
- assert False
- assert validInbox(bobDir,'bob',bobDomain)
- assert validInboxFilenames(bobDir,'bob',bobDomain,aliceDomain,alicePort)
-
- print('\n\n*********************************************************')
- print('Eve tries to send to Bob')
- sessionEve = createSession(eveDomain,evePort,useTor)
- eveSendThreads = []
- evePostLog = []
- evePersonCache={}
- eveCachedWebfingers={}
- eveSendThreads=[]
- evePostLog=[]
- useBlurhash=False
- sendResult = \
- sendPost(__version__, \
- sessionEve,eveDir,'eve', eveDomain, evePort, \
- 'bob', bobDomain, bobPort, ccUrl, \
- httpPrefix, 'Eve message', followersOnly, \
- saveToFile, clientToServer,None,None,None, \
- useBlurhash, federationList, 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(10):
- 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')
- print('\n\n*********************************************************')
- print('Alice sends a message to Bob')
- aliceSendThreads = []
- alicePostLog = []
- alicePersonCache={}
- aliceCachedWebfingers={}
- aliceSendThreads=[]
- alicePostLog=[]
- useBlurhash=False
- sendResult = \
- sendPost(__version__, \
- sessionAlice,aliceDir,'alice', aliceDomain, alicePort, \
- 'bob', bobDomain, bobPort, ccUrl, \
- httpPrefix, 'Alice message', followersOnly, saveToFile, \
- clientToServer,None,None,None,useBlurhash, federationList, \
- 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(20):
- 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))])>0:
- aliceMessageArrived=True
- print('Alice message sent to Bob!')
- break
- assert aliceMessageArrived==True
- print('Message from Alice to Bob succeeded, since it was granted capabilities')
- print('\n\n*********************************************************')
- print("\nBob changes Alice's capabilities so that she can't reply on his posts")
- bobCapsFilename= \
- bobDir+'/accounts/bob@'+bobDomain+'/ocap/accept/'+ \
- httpPrefix+':##'+aliceDomain+':'+str(alicePort)+'#users#alice.json'
- aliceCapsFilename= \
- aliceDir+'/accounts/alice@'+aliceDomain+'/ocap/granted/'+ \
- httpPrefix+':##'+bobDomain+':'+str(bobPort)+'#users#bob.json'
- sessionBob = createSession(bobDomain,bobPort,useTor)
- bobSendThreads = []
- bobPostLog = []
- bobPersonCache={}
- bobCachedWebfingers={}
- print("Bob's capabilities for Alice:")
- with open(bobCapsFilename, 'r') as fp:
- bobCapsJson=commentjson.load(fp)
- pprint(bobCapsJson)
- assert "inbox:noreply" not in bobCapsJson['capability']
- print("Alice's capabilities granted by Bob")
- with open(aliceCapsFilename, 'r') as fp:
- aliceCapsJson=commentjson.load(fp)
- pprint(aliceCapsJson)
- assert "inbox:noreply" not in aliceCapsJson['capability']
- newCapabilities=["inbox:write","objects:read","inbox:noreply"]
- sendCapabilitiesUpdate(sessionBob,bobDir,httpPrefix, \
- 'bob',bobDomain,bobPort, \
- httpPrefix+'://'+aliceDomain+':'+\
- str(alicePort)+'/users/alice',
- newCapabilities, \
- bobSendThreads, bobPostLog, \
- bobCachedWebfingers,bobPersonCache, \
- federationList,True,__version__)
- bobChanged=False
- bobNewCapsJson=None
- for i in range(20):
- time.sleep(1)
- with open(bobCapsFilename, 'r') as fp:
- bobNewCapsJson=commentjson.load(fp)
- if "inbox:noreply" in bobNewCapsJson['capability']:
- print("Bob's capabilities were changed")
- pprint(bobNewCapsJson)
- bobChanged=True
- break
- assert bobChanged
- aliceChanged=False
- aliceNewCapsJson=None
- for i in range(20):
- time.sleep(1)
- with open(aliceCapsFilename, 'r') as fp:
- aliceNewCapsJson=commentjson.load(fp)
- if "inbox:noreply" in aliceNewCapsJson['capability']:
- print("Alice's granted capabilities were changed")
- pprint(aliceNewCapsJson)
- aliceChanged=True
- break
- assert aliceChanged
- # check that the capabilities id has changed
- assert bobNewCapsJson['id']!=bobCapsJson['id']
- assert aliceNewCapsJson['id']!=aliceCapsJson['id']
- # stop the servers
- thrAlice.kill()
- thrAlice.join()
- assert thrAlice.isAlive()==False
- thrBob.kill()
- thrBob.join()
- assert thrBob.isAlive()==False
- thrEve.kill()
- thrEve.join()
- assert thrEve.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')
-
- assert 'alice@'+aliceDomain in open(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt').read()
- assert 'bob@'+bobDomain in open(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt').read()
- # 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(baseDir+'/.tests')
- 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)
- followPerson(baseDir,nickname,domain,'maxboardroom',domain,federationList,False)
- followPerson(baseDir,'drokk',domain,'ultrapancake',domain,federationList,False)
- # deliberate duplication
- followPerson(baseDir,'drokk',domain,'ultrapancake',domain,federationList,False)
- followPerson(baseDir,'sausagedog',domain,'ultrapancake',domain,federationList,False)
- followPerson(baseDir,nickname,domain,'ultrapancake',domain,federationList,False)
- followPerson(baseDir,nickname,domain,'someother','randodomain.net',federationList,False)
- 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)
- followPerson(baseDir,'drokk',otherdomain,nickname,domain,federationList,False)
- followPerson(baseDir,'sausagedog',otherdomain,nickname,domain,federationList,False)
- followPerson(baseDir,'maxboardroom',otherdomain,nickname,domain,federationList,False)
-
- followerOfPerson(baseDir,nickname,domain,'cucumber','sandwiches.party',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'captainsensible','damned.zone',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'pilchard','zombies.attack',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'drokk',otherdomain,federationList,False)
- followerOfPerson(baseDir,nickname,domain,'sausagedog',otherdomain,federationList,False)
- followerOfPerson(baseDir,nickname,domain,'maxboardroom',otherdomain,federationList,False)
- 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)
- def testGroupFollowers():
- print('testGroupFollowers')
- currDir=os.getcwd()
- nickname='test735'
- domain='mydomain.com'
- password='somepass'
- port=80
- httpPrefix='https'
- federationList=[]
- baseDir=currDir+'/.tests_testgroupfollowers'
- if os.path.isdir(baseDir):
- shutil.rmtree(baseDir)
- os.mkdir(baseDir)
- os.chdir(baseDir)
- createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
- clearFollowers(baseDir,nickname,domain)
- followerOfPerson(baseDir,nickname,domain,'badger','wild.domain',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'squirrel','wild.domain',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'rodent','wild.domain',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'utterly','clutterly.domain',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'zonked','zzz.domain',federationList,False)
- followerOfPerson(baseDir,nickname,domain,'nap','zzz.domain',federationList,False)
- grouped=groupFollowersByDomain(baseDir,nickname,domain)
- assert len(grouped.items())==3
- assert grouped.get('zzz.domain')
- assert grouped.get('clutterly.domain')
- assert grouped.get('wild.domain')
- assert len(grouped['zzz.domain'])==2
- assert len(grouped['wild.domain'])==3
- assert len(grouped['clutterly.domain'])==1
-
- os.chdir(currDir)
- shutil.rmtree(baseDir)
-
- def testFollows():
- print('testFollows')
- currDir=os.getcwd()
- nickname='test529'
- domain='testdomain.com'
- password='mypass'
- port=80
- httpPrefix='https'
- federationList=['wild.com','mesh.com']
- baseDir=currDir+'/.tests_testfollows'
- if os.path.isdir(baseDir):
- shutil.rmtree(baseDir)
- os.mkdir(baseDir)
- os.chdir(baseDir)
- createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
- clearFollows(baseDir,nickname,domain)
- 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)
- f = open(baseDir+'/accounts/'+nickname+'@'+domain+'/following.txt', "r")
- domainFound=False
- for followingDomain in f:
- testDomain=followingDomain.split('@')[1].replace('\n','')
- if testDomain=='mesh.com':
- domainFound=True
- if testDomain not in federationList:
- print(testDomain)
- assert(False)
- assert(domainFound)
- unfollowPerson(baseDir,nickname,domain,'batman','mesh.com')
- domainFound=False
- for followingDomain in f:
- testDomain=followingDomain.split('@')[1].replace('\n','')
- if testDomain=='mesh.com':
- domainFound=True
- assert(domainFound==False)
- clearFollowers(baseDir,nickname,domain)
- 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)
- f = open(baseDir+'/accounts/'+nickname+'@'+domain+'/followers.txt', "r")
- 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)
- def testCreatePerson():
- print('testCreatePerson')
- currDir=os.getcwd()
- nickname='test382'
- domain='badgerdomain.com'
- password='mypass'
- port=80
- httpPrefix='https'
- clientToServer=False
- useBlurhash=False
- baseDir=currDir+'/.tests_createperson'
- if os.path.isdir(baseDir):
- shutil.rmtree(baseDir)
- os.mkdir(baseDir)
- os.chdir(baseDir)
-
- privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
- assert os.path.isfile(baseDir+'/accounts/passwords')
- deleteAllPosts(baseDir,nickname,domain,'inbox')
- deleteAllPosts(baseDir,nickname,domain,'outbox')
- setDisplayNickname(baseDir,nickname,domain,'badger')
- setBio(baseDir,nickname,domain,'Randomly roaming in your backyard')
- archivePostsForPerson(nickname,domain,baseDir,'inbox',None,4)
- archivePostsForPerson(nickname,domain,baseDir,'outbox',None,4)
- createPublicPost(baseDir,nickname, domain, port,httpPrefix, "G'day world!", False, True, clientToServer,None,None,useBlurhash, None, None, 'Not suitable for Vogons')
- os.chdir(currDir)
- shutil.rmtree(baseDir)
- def testDelegateRoles():
- print('testDelegateRoles')
- currDir=os.getcwd()
- nickname='test382'
- nicknameDelegated='test383'
- domain='badgerdomain.com'
- password='mypass'
- port=80
- httpPrefix='https'
- clientToServer=False
- useBlurhash=False
- baseDir=currDir+'/.tests_delegaterole'
- if os.path.isdir(baseDir):
- shutil.rmtree(baseDir)
- os.mkdir(baseDir)
- os.chdir(baseDir)
-
- privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nickname,domain,port,httpPrefix,True,password)
- privateKeyPem,publicKeyPem,person,wfEndpoint=createPerson(baseDir,nicknameDelegated,domain,port,httpPrefix,True,'insecure')
- httpPrefix='http'
- project='artechoke'
- role='delegator'
- newRoleJson = {
- 'type': 'Delegate',
- 'actor': httpPrefix+'://'+domain+'/users/'+nickname,
- 'object': {
- 'type': 'Role',
- 'actor': httpPrefix+'://'+domain+'/users/'+nicknameDelegated,
- 'object': project+';'+role,
- 'to': [],
- 'cc': []
- },
- 'to': [],
- 'cc': []
- }
- assert outboxDelegate(baseDir,nickname,newRoleJson,False)
- # second time delegation has already happened so should return false
- assert outboxDelegate(baseDir,nickname,newRoleJson,False)==False
- assert '"delegator"' in open(baseDir+'/accounts/'+nickname+'@'+domain+'.json').read()
- assert '"delegator"' in open(baseDir+'/accounts/'+nicknameDelegated+'@'+domain+'.json').read()
-
- newRoleJson = {
- 'type': 'Delegate',
- 'actor': httpPrefix+'://'+domain+'/users/'+nicknameDelegated,
- 'object': {
- 'type': 'Role',
- 'actor': httpPrefix+'://'+domain+'/users/'+nickname,
- 'object': 'otherproject;otherrole',
- 'to': [],
- 'cc': []
- },
- 'to': [],
- 'cc': []
- }
- # non-delegators cannot assign roles
- assert outboxDelegate(baseDir,nicknameDelegated,newRoleJson,False)==False
- assert '"otherrole"' not in open(baseDir+'/accounts/'+nickname+'@'+domain+'.json').read()
- os.chdir(currDir)
- shutil.rmtree(baseDir)
- 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)
- 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
- authHeader=createBasicAuthHeader(nickname,password+'1')
- assert authorizeBasic(baseDir,'/users/'+nickname+'/inbox',authHeader,False)==False
- password='someOtherPassword'
- assert storeBasicCredentials(baseDir,nickname,password)
- authHeader=createBasicAuthHeader(nickname,password)
- assert authorizeBasic(baseDir,'/users/'+nickname+'/inbox',authHeader,False)
- os.chdir(currDir)
- shutil.rmtree(baseDir)
- def testClientToServer():
- print('Testing sending a post via c2s')
- global testServerAliceRunning
- global testServerBobRunning
- testServerAliceRunning = False
- testServerBobRunning = False
- httpPrefix='http'
- useTor=False
- federationList=[]
- baseDir=os.getcwd()
- if os.path.isdir(baseDir+'/.tests'):
- shutil.rmtree(baseDir+'/.tests')
- os.mkdir(baseDir+'/.tests')
- ocapAlways=False
- # create the servers
- aliceDir=baseDir+'/.tests/alice'
- aliceDomain='127.0.0.42'
- alicePort=61935
- thrAlice = \
- threadWithTrace(target=createServerAlice, \
- args=(aliceDir,aliceDomain,alicePort, \
- federationList,False,False, \
- ocapAlways),daemon=True)
-
- bobDir=baseDir+'/.tests/bob'
- bobDomain='127.0.0.64'
- bobPort=61936
- thrBob = \
- threadWithTrace(target=createServerBob, \
- args=(bobDir,bobDomain,bobPort, \
- federationList,False,False, \
- ocapAlways),daemon=True)
- thrAlice.start()
- thrBob.start()
- assert thrAlice.isAlive()==True
- assert thrBob.isAlive()==True
- # wait for both servers to be running
- ctr=0
- while not (testServerAliceRunning and testServerBobRunning):
- time.sleep(1)
- ctr+=1
- if ctr>60:
- break
- print('Alice online: '+str(testServerAliceRunning))
- print('Bob online: '+str(testServerBobRunning))
- time.sleep(1)
-
- print('\n\n*******************************************************')
- print('Alice sends to Bob via c2s')
- sessionAlice = createSession(aliceDomain,alicePort,useTor)
- followersOnly=False
- attachedImageFilename=baseDir+'/img/logo.png'
- mediaType=getAttachmentMediaType(attachedImageFilename)
- attachedImageDescription='Logo'
- useBlurhash=False
- cachedWebfingers={}
- personCache={}
- password='alicepass'
- outboxPath=aliceDir+'/accounts/alice@'+aliceDomain+'/outbox'
- inboxPath=bobDir+'/accounts/bob@'+bobDomain+'/inbox'
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==0
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==0
- sendResult= \
- sendPostViaServer(__version__, \
- aliceDir,sessionAlice,'alice',password, \
- aliceDomain,alicePort, \
- 'bob',bobDomain,bobPort,None, \
- httpPrefix,'Sent from my ActivityPub client',followersOnly, \
- attachedImageFilename,mediaType, \
- attachedImageDescription,useBlurhash, \
- cachedWebfingers,personCache, \
- True,None,None,None)
- print('sendResult: '+str(sendResult))
- for i in range(30):
- if os.path.isdir(outboxPath):
- if len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==1:
- break
- time.sleep(1)
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==1
- print(">>> c2s post arrived in Alice's outbox")
-
- for i in range(30):
- if os.path.isdir(inboxPath):
- if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1:
- break
- time.sleep(1)
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1
- print(">>> s2s post arrived in Bob's inbox")
- print("c2s send success")
- print('\n\nGetting message id for the post')
- statusNumber=0
- outboxPostFilename=None
- outboxPostId=None
- for name in os.listdir(outboxPath):
- if '#statuses#' in name:
- statusNumber=int(name.split('#statuses#')[1].replace('.json','').replace('#activity',''))
- outboxPostFilename=outboxPath+'/'+name
- with open(outboxPostFilename, 'r') as fp:
- postJsonObject=commentjson.load(fp)
- outboxPostId=postJsonObject['id'].replace('/activity','')
- assert outboxPostId
- print('message id obtained: '+outboxPostId)
- assert validInbox(bobDir,'bob',bobDomain)
- assert validInboxFilenames(bobDir,'bob',bobDomain,aliceDomain,alicePort)
- print('\n\nAlice follows Bob')
- sendFollowRequestViaServer(aliceDir,sessionAlice, \
- 'alice',password, \
- aliceDomain,alicePort, \
- 'bob',bobDomain,bobPort, \
- httpPrefix, \
- cachedWebfingers,personCache, \
- True,__version__)
- for t in range(10):
- if os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt'):
- if 'alice@'+aliceDomain+':'+str(alicePort) in open(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt').read():
- if os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt'):
- if 'bob@'+bobDomain+':'+str(bobPort) in open(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt').read():
- break
- time.sleep(1)
- assert os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt')
- assert os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt')
- assert 'alice@'+aliceDomain+':'+str(alicePort) in open(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt').read()
- assert 'bob@'+bobDomain+':'+str(bobPort) in open(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt').read()
- assert validInbox(bobDir,'bob',bobDomain)
- assert validInboxFilenames(bobDir,'bob',bobDomain,aliceDomain,alicePort)
- print('\n\nBob follows Alice')
- sendFollowRequestViaServer(aliceDir,sessionAlice, \
- 'bob','bobpass', \
- bobDomain,bobPort, \
- 'alice',aliceDomain,alicePort, \
- httpPrefix, \
- cachedWebfingers,personCache, \
- True,__version__)
- for t in range(10):
- if os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/followers.txt'):
- if 'bob@'+bobDomain+':'+str(bobPort) in open(aliceDir+'/accounts/alice@'+aliceDomain+'/followers.txt').read():
- if os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/following.txt'):
- if 'alice@'+aliceDomain+':'+str(alicePort) in open(bobDir+'/accounts/bob@'+bobDomain+'/following.txt').read():
- break
- time.sleep(1)
- assert os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/followers.txt')
- assert os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/following.txt')
- assert 'bob@'+bobDomain+':'+str(bobPort) in open(aliceDir+'/accounts/alice@'+aliceDomain+'/followers.txt').read()
- assert 'alice@'+aliceDomain+':'+str(alicePort) in open(bobDir+'/accounts/bob@'+bobDomain+'/following.txt').read()
- print('\n\nBob likes the post')
- sessionBob = createSession(bobDomain,bobPort,useTor)
- password='bobpass'
- outboxPath=bobDir+'/accounts/bob@'+bobDomain+'/outbox'
- inboxPath=aliceDir+'/accounts/alice@'+aliceDomain+'/inbox'
- print(str(len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])))
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==1
- print(str(len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])))
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1
- sendLikeViaServer(bobDir,sessionBob, \
- 'bob','bobpass', \
- bobDomain,bobPort, \
- httpPrefix,outboxPostId, \
- cachedWebfingers,personCache, \
- True,__version__)
- for i in range(20):
- if os.path.isdir(outboxPath) and os.path.isdir(inboxPath):
- if len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==2:
- if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1:
- break
- time.sleep(1)
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==2
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1
- print('Post liked')
-
- print('\n\nBob repeats the post')
- print(str(len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])))
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==2
- print(str(len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])))
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==1
- sendAnnounceViaServer(bobDir,sessionBob,'bob',password, \
- bobDomain,bobPort, \
- httpPrefix,outboxPostId, \
- cachedWebfingers, \
- personCache,True,__version__)
- for i in range(20):
- if os.path.isdir(outboxPath) and os.path.isdir(inboxPath):
- if len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==3:
- if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==2:
- break
- time.sleep(1)
- assert len([name for name in os.listdir(outboxPath) if os.path.isfile(os.path.join(outboxPath, name))])==3
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==2
- print('Post repeated')
- inboxPath=bobDir+'/accounts/bob@'+bobDomain+'/inbox'
- outboxPath=aliceDir+'/accounts/alice@'+aliceDomain+'/outbox'
- postsBefore = len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])
- print('\n\nAlice deletes her post: '+outboxPostId+' '+str(postsBefore))
- password='alicepass'
- sendDeleteViaServer(aliceDir,sessionAlice,'alice',password,
- aliceDomain,alicePort, \
- httpPrefix,outboxPostId, \
- cachedWebfingers,personCache, \
- True,__version__)
- for i in range(30):
- if os.path.isdir(inboxPath):
- if len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==postsBefore-1:
- break
- time.sleep(1)
- assert len([name for name in os.listdir(inboxPath) if os.path.isfile(os.path.join(inboxPath, name))])==postsBefore-1
- print(">>> post deleted from Alice's outbox and Bob's inbox")
- assert validInbox(bobDir,'bob',bobDomain)
- assert validInboxFilenames(bobDir,'bob',bobDomain,aliceDomain,alicePort)
-
- print('\n\nAlice unfollows Bob')
- password='alicepass'
- sendUnfollowRequestViaServer(baseDir,sessionAlice, \
- 'alice',password, \
- aliceDomain,alicePort, \
- 'bob',bobDomain,bobPort, \
- httpPrefix, \
- cachedWebfingers,personCache, \
- True,__version__)
- for t in range(10):
- if 'alice@'+aliceDomain+':'+str(alicePort) not in open(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt').read():
- if 'bob@'+bobDomain+':'+str(bobPort) not in open(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt').read():
- break
- time.sleep(1)
- assert os.path.isfile(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt')
- assert os.path.isfile(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt')
- assert 'alice@'+aliceDomain+':'+str(alicePort) not in open(bobDir+'/accounts/bob@'+bobDomain+'/followers.txt').read()
- assert 'bob@'+bobDomain+':'+str(bobPort) not in open(aliceDir+'/accounts/alice@'+aliceDomain+'/following.txt').read()
- assert validInbox(bobDir,'bob',bobDomain)
- assert validInboxFilenames(bobDir,'bob',bobDomain,aliceDomain,alicePort)
- assert validInbox(aliceDir,'alice',aliceDomain)
- assert validInboxFilenames(aliceDir,'alice',aliceDomain,bobDomain,bobPort)
- # stop the servers
- thrAlice.kill()
- thrAlice.join()
- assert thrAlice.isAlive()==False
- thrBob.kill()
- thrBob.join()
- assert thrBob.isAlive()==False
- os.chdir(baseDir)
- #shutil.rmtree(aliceDir)
- #shutil.rmtree(bobDir)
- def testActorParsing():
- print('testActorParsing')
- actor='https://mydomain:72/users/mynick'
- domain,port=getDomainFromActor(actor)
- assert domain=='mydomain'
- assert port==72
- nickname=getNicknameFromActor(actor)
- assert nickname=='mynick'
- actor='https://randomain/users/rando'
- domain,port=getDomainFromActor(actor)
- assert domain=='randomain'
- nickname=getNicknameFromActor(actor)
- assert nickname=='rando'
- actor='https://otherdomain:49/@othernick'
- domain,port=getDomainFromActor(actor)
- assert domain=='otherdomain'
- assert port==49
- nickname=getNicknameFromActor(actor)
- assert nickname=='othernick'
- def testWebLinks():
- print('testWebLinks')
- exampleText='This post has a web links https://somesite.net\n\nAnd some other text'
- linkedText=addWebLinks(exampleText)
- assert '<a href="https://somesite.net" rel="nofollow noopener" target="_blank"><span class="invisible">https://</span><span class="ellipsis">somesite.net</span></a' in linkedText
-
- def runAllTests():
- print('Running tests...')
- testWebLinks()
- testActorParsing()
- testHttpsig()
- testCache()
- testThreads()
- testCreatePerson()
- testAuthentication()
- testFollowersOfPerson()
- testNoOfFollowersOnDomain()
- testFollows()
- testGroupFollowers()
- testDelegateRoles()
- print('Tests succeeded\n')
|