123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989 |
- # -*- coding: utf-8 -*-
- """
- (c) 2015-2017 - Copyright Red Hat Inc
- Authors:
- Pierre-Yves Chibon <pingou@pingoured.fr>
- """
- from __future__ import unicode_literals
- __requires__ = ['SQLAlchemy >= 0.8']
- import pkg_resources
- import datetime
- import json
- import unittest
- import shutil
- import sys
- import time
- import os
- import pygit2
- from mock import patch
- sys.path.insert(0, os.path.join(os.path.dirname(
- os.path.abspath(__file__)), '..'))
- import pagure
- import pagure.lib
- import tests
- from pagure.lib.repo import PagureRepo
- class PagureFlaskInternaltests(tests.Modeltests):
- """ Tests for flask Internal controller of pagure """
- def setUp(self):
- """ Set up the environnment, ran before every tests. """
- super(PagureFlaskInternaltests, self).setUp()
- pagure.config.config['IP_ALLOWED_INTERNAL'] = list(set(
- pagure.config.config['IP_ALLOWED_INTERNAL'] + [None]))
- pagure.config.config['GIT_FOLDER'] = os.path.join(
- self.path, 'repos')
- pagure.config.config['REQUESTS_FOLDER'] = None
- pagure.config.config['TICKETS_FOLDER'] = None
- pagure.config.config['DOCS_FOLDER'] = None
- @patch('pagure.lib.notify.send_email')
- def test_pull_request_add_comment(self, send_email):
- """ Test the pull_request_add_comment function. """
- send_email.return_value = True
- tests.create_projects(self.session)
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- req = pagure.lib.new_pull_request(
- session=self.session,
- repo_from=repo,
- branch_from='feature',
- repo_to=repo,
- branch_to='master',
- title='PR from the feature branch',
- user='pingou',
- requestfolder=None,
- )
- self.session.commit()
- self.assertEqual(req.id, 1)
- self.assertEqual(req.title, 'PR from the feature branch')
- request = repo.requests[0]
- self.assertEqual(len(request.comments), 0)
- self.assertEqual(len(request.discussion), 0)
- data = {
- 'objid': 'foo',
- }
- # Wrong http request
- output = self.app.post('/pv/pull-request/comment/', data=data)
- self.assertEqual(output.status_code, 405)
- # Invalid request
- output = self.app.put('/pv/pull-request/comment/', data=data)
- self.assertEqual(output.status_code, 400)
- data = {
- 'objid': 'foo',
- 'useremail': 'foo@pingou.com',
- }
- # Invalid objid
- output = self.app.put('/pv/pull-request/comment/', data=data)
- self.assertEqual(output.status_code, 404)
- data = {
- 'objid': request.uid,
- 'useremail': 'foo@pingou.com',
- }
- # Valid objid, in-complete data for a comment
- output = self.app.put('/pv/pull-request/comment/', data=data)
- self.assertEqual(output.status_code, 400)
- data = {
- 'objid': request.uid,
- 'useremail': 'foo@pingou.com',
- 'comment': 'Looks good to me!',
- }
- # Add comment
- output = self.app.put('/pv/pull-request/comment/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, {'message': 'Comment added'})
- self.session.commit()
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- request = repo.requests[0]
- self.assertEqual(len(request.comments), 1)
- self.assertEqual(len(request.discussion), 1)
- # Check the @localonly
- before = pagure.config.config['IP_ALLOWED_INTERNAL'][:]
- pagure.config.config['IP_ALLOWED_INTERNAL'] = []
- output = self.app.put('/pv/pull-request/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- pagure.config.config['IP_ALLOWED_INTERNAL'] = before[:]
- @patch('pagure.lib.notify.send_email')
- def test_ticket_add_comment(self, send_email):
- """ Test the ticket_add_comment function. """
- send_email.return_value = True
- tests.create_projects(self.session)
- # Create issues to play with
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- msg = pagure.lib.new_issue(
- session=self.session,
- repo=repo,
- title='Test issue',
- content='We should work on this',
- user='pingou',
- ticketfolder=None
- )
- self.session.commit()
- self.assertEqual(msg.title, 'Test issue')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 0)
- data = {
- 'objid': 'foo',
- }
- # Wrong http request
- output = self.app.post('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 405)
- # Invalid request
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 400)
- data = {
- 'objid': 'foo',
- 'useremail': 'foo@pingou.com',
- }
- # Invalid objid
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 404)
- data = {
- 'objid': issue.uid,
- 'useremail': 'foo@pingou.com',
- }
- # Valid objid, in-complete data for a comment
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 400)
- data = {
- 'objid': issue.uid,
- 'useremail': 'foo@pingou.com',
- 'comment': 'Looks good to me!',
- }
- # Add comment
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, {'message': 'Comment added'})
- self.session.commit()
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 1)
- # Check the @localonly
- pagure.config.config['IP_ALLOWED_INTERNAL'].remove(None)
- before = pagure.config.config['IP_ALLOWED_INTERNAL'][:]
- pagure.config.config['IP_ALLOWED_INTERNAL'] = []
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- pagure.config.config['IP_ALLOWED_INTERNAL'] = before[:]
- @patch('pagure.lib.notify.send_email')
- def test_private_ticket_add_comment(self, send_email):
- """ Test the ticket_add_comment function on a private ticket. """
- send_email.return_value = True
- tests.create_projects(self.session)
- # Create issues to play with
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- msg = pagure.lib.new_issue(
- session=self.session,
- repo=repo,
- title='Test issue',
- content='We should work on this, really',
- user='pingou',
- private=True,
- ticketfolder=None
- )
- self.session.commit()
- self.assertEqual(msg.title, 'Test issue')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 0)
- data = {
- 'objid': 'foo',
- }
- # Wrong http request
- output = self.app.post('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 405)
- # Invalid request
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 400)
- data = {
- 'objid': 'foo',
- 'useremail': 'foo@pingou.com',
- }
- # Invalid objid
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 404)
- data = {
- 'objid': issue.uid,
- 'useremail': 'foo@bar.com',
- }
- # Valid objid, un-allowed user for this (private) ticket
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- data = {
- 'objid': issue.uid,
- 'useremail': 'foo@pingou.com',
- }
- # Valid objid, un-allowed user for this (private) ticket
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 400)
- data = {
- 'objid': issue.uid,
- 'useremail': 'foo@pingou.com',
- 'comment': 'Looks good to me!',
- }
- # Add comment
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, {'message': 'Comment added'})
- self.session.commit()
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 1)
- # Check the @localonly
- before = pagure.config.config['IP_ALLOWED_INTERNAL'][:]
- pagure.config.config['IP_ALLOWED_INTERNAL'] = []
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- pagure.config.config['IP_ALLOWED_INTERNAL'] = before[:]
- @patch('pagure.lib.notify.send_email')
- def test_private_ticket_add_comment_acl(self, send_email):
- """ Test the ticket_add_comment function on a private ticket. """
- send_email.return_value = True
- tests.create_projects(self.session)
- # Create issues to play with
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- msg = pagure.lib.new_issue(
- session=self.session,
- repo=repo,
- title='Test issue',
- content='We should work on this, really',
- user='pingou',
- private=True,
- ticketfolder=None
- )
- self.session.commit()
- self.assertEqual(msg.title, 'Test issue')
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 0)
- # Currently, he is just an average user,
- # He doesn't have any access in this repo
- data = {
- 'objid': issue.uid,
- 'useremail': 'foo@bar.com',
- 'comment': 'Looks good to me!',
- }
- # Valid objid, un-allowed user for this (private) ticket
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- # Let's promote him to be a ticketer
- # He shoudn't be able to comment even then though
- msg = pagure.lib.add_user_to_project(
- self.session,
- project=repo,
- new_user='foo',
- user='pingou',
- access='ticket'
- )
- self.session.commit()
- self.assertEqual(msg, 'User added')
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- self.assertEqual(
- sorted([u.username for u in repo.users]), ['foo'])
- self.assertEqual(
- sorted([u.username for u in repo.committers]), [])
- self.assertEqual(
- sorted([u.username for u in repo.admins]), [])
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- # Let's promote him to be a committer
- # He should be able to comment
- msg = pagure.lib.add_user_to_project(
- self.session,
- project=repo,
- new_user='foo',
- user='pingou',
- access='commit'
- )
- self.session.commit()
- self.assertEqual(msg, 'User access updated')
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- self.assertEqual(
- sorted([u.username for u in repo.users]), ['foo'])
- self.assertEqual(
- sorted([u.username for u in repo.committers]), ['foo'])
- self.assertEqual(
- sorted([u.username for u in repo.admins]), [])
- # Add comment
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, {'message': 'Comment added'})
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 1)
- # Let's promote him to be a admin
- # He should be able to comment
- msg = pagure.lib.add_user_to_project(
- self.session,
- project=repo,
- new_user='foo',
- user='pingou',
- access='admin'
- )
- self.session.commit()
- self.assertEqual(msg, 'User access updated')
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- self.assertEqual(
- sorted([u.username for u in repo.users]), ['foo'])
- self.assertEqual(
- sorted([u.username for u in repo.committers]), ['foo'])
- self.assertEqual(
- sorted([u.username for u in repo.admins]), ['foo'])
- # Add comment
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, {'message': 'Comment added'})
- repo = pagure.lib.get_authorized_project(self.session, 'test')
- issue = repo.issues[0]
- self.assertEqual(len(issue.comments), 2)
- # Check the @localonly
- before = pagure.config.config['IP_ALLOWED_INTERNAL'][:]
- pagure.config.config['IP_ALLOWED_INTERNAL'] = []
- output = self.app.put('/pv/ticket/comment/', data=data)
- self.assertEqual(output.status_code, 403)
- pagure.config.config['IP_ALLOWED_INTERNAL'] = before[:]
- @patch('pagure.lib.notify.send_email')
- def test_mergeable_request_pull_FF(self, send_email):
- """ Test the mergeable_request_pull endpoint with a fast-forward
- merge.
- """
- send_email.return_value = True
- # Create a git repo to play with
- origgitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertFalse(os.path.exists(origgitrepo))
- os.makedirs(origgitrepo)
- orig_repo = pygit2.init_repository(origgitrepo, bare=True)
- os.makedirs(os.path.join(self.path, 'repos_tmp'))
- gitrepo = os.path.join(self.path, 'repos_tmp', 'test.git')
- repo = pygit2.clone_repository(origgitrepo, gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- refname = 'refs/heads/master:refs/heads/master'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/feature', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- second_commit = repo.revparse_single('HEAD')
- refname = 'refs/heads/feature:refs/heads/feature'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Create a PR for these changes
- tests.create_projects(self.session)
- project = pagure.lib.get_authorized_project(self.session, 'test')
- req = pagure.lib.new_pull_request(
- session=self.session,
- repo_from=project,
- branch_from='feature',
- repo_to=project,
- branch_to='master',
- title='PR from the feature branch',
- user='pingou',
- requestfolder=None,
- )
- self.session.commit()
- self.assertEqual(req.id, 1)
- self.assertEqual(req.title, 'PR from the feature branch')
- # Check if the PR can be merged
- data = {
- 'objid': 'blah',
- }
- # Missing CSRF
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 400)
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- output = self.app.get('/test/adduser')
- csrf_token = output.get_data(as_text=True).split(
- 'name="csrf_token" type="hidden" value="')[1].split('">')[0]
- # Missing request identifier
- data = {
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 404)
- # With all the desired information
- project = pagure.lib.get_authorized_project(self.session, 'test')
- data = {
- 'csrf_token': csrf_token,
- 'requestid': project.requests[0].uid,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 200)
- exp = {
- "code": "FFORWARD",
- "message": "The pull-request can be merged and fast-forwarded",
- "short_code": "Ok"
- }
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, exp)
- @patch('pagure.lib.notify.send_email')
- def test_mergeable_request_pull_no_change(self, send_email):
- """ Test the mergeable_request_pull endpoint when there are no
- changes to merge.
- """
- send_email.return_value = True
- # Create a git repo to play with
- gitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertFalse(os.path.exists(gitrepo))
- os.makedirs(gitrepo)
- repo = pygit2.init_repository(gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- second_commit = repo.revparse_single('HEAD')
- # Create a PR for these changes
- tests.create_projects(self.session)
- project = pagure.lib.get_authorized_project(self.session, 'test')
- req = pagure.lib.new_pull_request(
- session=self.session,
- repo_from=project,
- branch_from='master',
- repo_to=project,
- branch_to='master',
- title='PR from the feature branch',
- user='pingou',
- requestfolder=None,
- )
- self.session.commit()
- self.assertEqual(req.id, 1)
- self.assertEqual(req.title, 'PR from the feature branch')
- # Check if the PR can be merged
- data = {
- 'objid': 'blah',
- }
- # Missing CSRF
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 400)
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- output = self.app.get('/test/adduser')
- csrf_token = output.get_data(as_text=True).split(
- 'name="csrf_token" type="hidden" value="')[1].split('">')[0]
- # Missing request identifier
- data = {
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 404)
- # With all the desired information
- project = pagure.lib.get_authorized_project(self.session, 'test')
- data = {
- 'csrf_token': csrf_token,
- 'requestid': project.requests[0].uid,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 200)
- exp = {
- "code": "NO_CHANGE",
- "message": "Nothing to change, git is up to date",
- "short_code": "No changes"
- }
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, exp)
- @patch('pagure.lib.notify.send_email')
- def test_mergeable_request_pull_merge(self, send_email):
- """ Test the mergeable_request_pull endpoint when the changes can
- be merged with a merge commit.
- """
- send_email.return_value = True
- # Create a git repo to play with
- origgitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertFalse(os.path.exists(origgitrepo))
- os.makedirs(origgitrepo)
- orig_repo = pygit2.init_repository(origgitrepo, bare=True)
- os.makedirs(os.path.join(self.path, 'repos_tmp'))
- gitrepo = os.path.join(self.path, 'repos_tmp', 'test.git')
- repo = pygit2.clone_repository(origgitrepo, gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- refname = 'refs/heads/master:refs/heads/master'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/feature', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- refname = 'refs/heads/feature:refs/heads/feature'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Create another file in the master branch
- with open(os.path.join(gitrepo, '.gitignore'), 'w') as stream:
- stream.write('*~')
- repo.index.add('.gitignore')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add .gitignore file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- refname = 'refs/heads/master:refs/heads/master'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Create a PR for these changes
- tests.create_projects(self.session)
- project = pagure.lib.get_authorized_project(self.session, 'test')
- req = pagure.lib.new_pull_request(
- session=self.session,
- repo_from=project,
- branch_from='feature',
- repo_to=project,
- branch_to='master',
- title='PR from the feature branch',
- user='pingou',
- requestfolder=None,
- )
- self.session.commit()
- self.assertEqual(req.id, 1)
- self.assertEqual(req.title, 'PR from the feature branch')
- # Check if the PR can be merged
- data = {}
- # Missing CSRF
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 400)
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- output = self.app.get('/test/adduser')
- csrf_token = output.get_data(as_text=True).split(
- 'name="csrf_token" type="hidden" value="')[1].split('">')[0]
- # Missing request identifier
- data = {
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 404)
- # With all the desired information
- project = pagure.lib.get_authorized_project(self.session, 'test')
- data = {
- 'csrf_token': csrf_token,
- 'requestid': project.requests[0].uid,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 200)
- exp = {
- "code": "MERGE",
- "message": "The pull-request can be merged with a merge commit",
- "short_code": "With merge"
- }
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, exp)
- @patch('pagure.lib.notify.send_email')
- def test_mergeable_request_pull_conflicts(self, send_email):
- """ Test the mergeable_request_pull endpoint when the changes cannot
- be merged due to conflicts.
- """
- send_email.return_value = True
- # Create a git repo to play with
- origgitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertFalse(os.path.exists(origgitrepo))
- os.makedirs(origgitrepo)
- orig_repo = pygit2.init_repository(origgitrepo, bare=True)
- os.makedirs(os.path.join(self.path, 'repos_tmp'))
- gitrepo = os.path.join(self.path, 'repos_tmp', 'test.git')
- repo = pygit2.clone_repository(origgitrepo, gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- refname = 'refs/heads/master:refs/heads/master'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/feature', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- refname = 'refs/heads/feature:refs/heads/feature'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Create another file in the master branch
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add .gitignore file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- refname = 'refs/heads/master:refs/heads/master'
- ori_remote = repo.remotes[0]
- PagureRepo.push(ori_remote, refname)
- # Create a PR for these changes
- tests.create_projects(self.session)
- project = pagure.lib.get_authorized_project(self.session, 'test')
- req = pagure.lib.new_pull_request(
- session=self.session,
- repo_from=project,
- branch_from='feature',
- repo_to=project,
- branch_to='master',
- title='PR from the feature branch',
- user='pingou',
- requestfolder=None,
- )
- self.session.commit()
- self.assertEqual(req.id, 1)
- self.assertEqual(req.title, 'PR from the feature branch')
- # Check if the PR can be merged
- data = {}
- # Missing CSRF
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 400)
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- output = self.app.get('/test/adduser')
- csrf_token = output.get_data(as_text=True).split(
- 'name="csrf_token" type="hidden" value="')[1].split('">')[0]
- # Missing request identifier
- data = {
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 404)
- # With all the desired information
- project = pagure.lib.get_authorized_project(self.session, 'test')
- data = {
- 'csrf_token': csrf_token,
- 'requestid': project.requests[0].uid,
- }
- output = self.app.post('/pv/pull-request/merge', data=data)
- self.assertEqual(output.status_code, 200)
- exp = {
- "code": "CONFLICTS",
- "message": "The pull-request cannot be merged due to conflicts",
- "short_code": "Conflicts"
- }
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(js_data, exp)
- def test_get_branches_of_commit(self):
- ''' Test the get_branches_of_commit from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(os.path.join(self.path, 'repos'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- output = self.app.get('/test/adduser')
- self.assertEqual(output.status_code, 200)
- csrf_token = output.get_data(as_text=True).split(
- 'name="csrf_token" type="hidden" value="')[1].split('">')[0]
- # No CSRF token
- data = {
- 'repo': 'fakerepo',
- 'commit_id': 'foo',
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR', 'message': 'Invalid input submitted'}
- )
- # Invalid repo
- data = {
- 'repo': 'fakerepo',
- 'commit_id': 'foo',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- 'code': 'ERROR',
- 'message': 'No repo found with the information provided'
- }
- )
- # Rigth repo, no commit
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR', 'message': 'No commit id submitted'}
- )
- # Request is fine, but git repo doesn't exist
- item = pagure.lib.model.Project(
- user_id=1, # pingou
- name='test20',
- description='test project #20',
- hook_token='aaabbbhhh',
- )
- self.session.add(item)
- self.session.commit()
- data = {
- 'repo': 'test20',
- 'commit_id': 'foo',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- 'code': 'ERROR',
- 'message': 'No git repo found with the information provided'
- }
- )
- # Create a git repo to play with
- gitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertTrue(os.path.exists(gitrepo))
- repo = pygit2.Repository(gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/feature', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- # Create another file in the master branch
- with open(os.path.join(gitrepo, '.gitignore'), 'w') as stream:
- stream.write('*~')
- repo.index.add('.gitignore')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- commit_hash = repo.create_commit(
- 'refs/heads/feature_branch', # the name of the reference to update
- author,
- committer,
- 'Add .gitignore file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- # All good but the commit id
- data = {
- 'repo': 'test',
- 'commit_id': 'foo',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- 'code': 'ERROR',
- 'message': 'This commit could not be found in this repo'
- }
- )
- # All good
- data = {
- 'repo': 'test',
- 'commit_id': commit_hash,
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- 'code': 'OK',
- 'branches': ['feature_branch'],
- }
- )
- def test_get_branches_of_commit_with_unrelated_branches(self):
- ''' Test the get_branches_of_commit from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(os.path.join(self.path, 'repos'))
- user = tests.FakeUser(username='pingou')
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Create a git repo to play with
- gitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertTrue(os.path.exists(gitrepo))
- repo = pygit2.Repository(gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added, but unrelated with the first commit
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- commit = repo.create_commit(
- 'refs/heads/feature', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- commit_hash = commit.hex
- # All good
- data = {
- 'repo': 'test',
- 'commit_id': commit_hash,
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/commit/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- u'code': u'OK',
- u'branches': ['feature'],
- }
- )
- def test_get_branches_head(self):
- ''' Test the get_branches_head from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(os.path.join(self.path, 'repos'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # No CSRF token
- data = {
- 'repo': 'fakerepo',
- }
- output = self.app.post('/pv/branches/heads/', data=data)
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR', 'message': 'Invalid input submitted'}
- )
- # Invalid repo
- data = {
- 'repo': 'fakerepo',
- 'commit_id': 'foo',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/heads/', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- 'code': 'ERROR',
- 'message': 'No repo found with the information provided'
- }
- )
- # Rigth repo, no commit
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/heads/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {"branches": {}, "code": "OK", "heads": {}}
- )
- # Request is fine, but git repo doesn't exist
- item = pagure.lib.model.Project(
- user_id=1, # pingou
- name='test20',
- description='test project #20',
- hook_token='aaabbbhhh',
- )
- self.session.add(item)
- self.session.commit()
- data = {
- 'repo': 'test20',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/heads/', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {
- 'code': 'ERROR',
- 'message': 'No git repo found with the information provided'
- }
- )
- # Create a git repo to play with
- gitrepo = os.path.join(self.path, 'repos', 'test.git')
- self.assertTrue(os.path.exists(gitrepo))
- repo = pygit2.Repository(gitrepo)
- # Create a file in that git repo
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/master', # the name of the reference to update
- author,
- committer,
- 'Add sources file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- []
- )
- first_commit = repo.revparse_single('HEAD')
- # Edit the sources file again
- with open(os.path.join(gitrepo, 'sources'), 'w') as stream:
- stream.write('foo\n bar\nbaz\n boose')
- repo.index.add('sources')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- repo.create_commit(
- 'refs/heads/feature', # the name of the reference to update
- author,
- committer,
- 'Add baz and boose to the sources\n\n There are more objects to '
- 'consider',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- # Create another file in the master branch
- with open(os.path.join(gitrepo, '.gitignore'), 'w') as stream:
- stream.write('*~')
- repo.index.add('.gitignore')
- repo.index.write()
- # Commits the files added
- tree = repo.index.write_tree()
- author = pygit2.Signature(
- 'Alice Author', 'alice@authors.tld')
- committer = pygit2.Signature(
- 'Cecil Committer', 'cecil@committers.tld')
- commit_hash = repo.create_commit(
- 'refs/heads/feature_branch', # the name of the reference to update
- author,
- committer,
- 'Add .gitignore file for testing',
- # binary string representing the tree object ID
- tree,
- # list of binary strings representing parents of the new commit
- [first_commit.oid.hex]
- )
- # All good
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/branches/heads/', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- # We can't test the content since the commit hash will change all
- # the time, so let's just check the structure
- self.assertEqual(
- sorted(js_data.keys()), ['branches', 'code', 'heads'])
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(len(js_data['heads']), 3)
- self.assertEqual(len(js_data['branches']), 3)
- def test_get_stats_commits_no_token(self):
- ''' Test the get_stats_commits from the internal API. '''
- # No CSRF token
- data = {
- 'repo': 'fakerepo',
- }
- output = self.app.post('/pv/stats/commits/authors', data=data)
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR', 'message': 'Invalid input submitted'}
- )
- def test_get_stats_commits_invalid_repo(self):
- ''' Test the get_stats_commits from the internal API. '''
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Invalid repo
- data = {
- 'repo': 'fakerepo',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/stats/commits/authors', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR',
- 'message': 'No repo found with the information provided'}
- )
- def test_get_stats_commits_empty_git(self):
- ''' Test the get_stats_commits from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(os.path.join(self.path, 'repos'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # No content in git
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/stats/commits/authors', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message', 'task_id', 'url']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(js_data['message'], 'Stats asked')
- self.assertTrue(js_data['url'].startswith('/pv/task/'))
- output = self.app.get(js_data['url'])
- js_data2 = json.loads(output.get_data(as_text=True))
- self.assertTrue(
- js_data2 in [
- {'results': "reference 'refs/heads/master' not found"},
- {'results': "Reference 'refs/heads/master' not found"}
- ]
- )
- def test_get_stats_commits_git_populated(self):
- ''' Test the get_stats_commits from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'test.git'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Content in git
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/stats/commits/authors', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message', 'task_id', 'url']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(js_data['message'], 'Stats asked')
- self.assertTrue(js_data['url'].startswith('/pv/task/'))
- output = self.app.get(js_data['url'])
- while output.status_code == 418:
- time.sleep(0.5)
- output = self.app.get(js_data['url'])
- js_data2 = json.loads(output.get_data(as_text=True))
- self.assertTrue(js_data2['results'][3] > 1509110062)
- js_data2['results'][3] = 1509110062
- self.assertDictEqual(
- js_data2,
- {'results': [
- 2,
- [[2, [['Alice Author', 'alice@authors.tld']]]],
- 1,
- 1509110062
- ]
- }
- )
- def test_get_stats_commits_trend_no_token(self):
- ''' Test the get_stats_commits_trend from the internal API. '''
- # No CSRF token
- data = {
- 'repo': 'fakerepo',
- }
- output = self.app.post('/pv/stats/commits/trend', data=data)
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR', 'message': 'Invalid input submitted'}
- )
- def test_get_stats_commits_trend_invalid_repo(self):
- """ Test the get_stats_commits_trend from the internal API. """
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Invalid repo
- data = {
- 'repo': 'fakerepo',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/stats/commits/trend', data=data)
- self.assertEqual(output.status_code, 404)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertDictEqual(
- js_data,
- {'code': 'ERROR',
- 'message': 'No repo found with the information provided'}
- )
- def test_get_stats_commits_trend_empty_git(self):
- ''' Test the get_stats_commits_trend from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(os.path.join(self.path, 'repos'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # No content in git
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/stats/commits/trend', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message', 'task_id', 'url']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(js_data['message'], 'Stats asked')
- self.assertTrue(js_data['url'].startswith('/pv/task/'))
- output = self.app.get(js_data['url'])
- js_data2 = json.loads(output.get_data(as_text=True))
- self.assertTrue(
- js_data2 in [
- {'results': "reference 'refs/heads/master' not found"},
- {'results': "Reference 'refs/heads/master' not found"}
- ]
- )
- def test_get_stats_commits_trend_git_populated(self):
- ''' Test the get_stats_commits_trend from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'test.git'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Content in git
- data = {
- 'repo': 'test',
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/stats/commits/trend', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message', 'task_id', 'url']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(js_data['message'], 'Stats asked')
- self.assertTrue(js_data['url'].startswith('/pv/task/'))
- output = self.app.get(js_data['url'])
- js_data2 = json.loads(output.get_data(as_text=True))
- today = datetime.datetime.utcnow().date()
- self.assertDictEqual(
- js_data2,
- {'results': [[str(today), 2]]}
- )
- def test_get_project_family_no_project(self):
- ''' Test the get_project_family from the internal API. '''
- output = self.app.post('/pv/test/family')
- self.assertEqual(output.status_code, 404)
- def test_get_project_family_no_csrf(self):
- ''' Test the get_project_family from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'test.git'))
- output = self.app.post('/pv/test/family')
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message']
- )
- self.assertEqual(js_data['code'], 'ERROR')
- self.assertEqual(js_data['message'], 'Invalid input submitted')
- def test_get_project_family(self):
- ''' Test the get_project_family from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'test.git'))
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- data = {
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/test/family', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'family']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(js_data['family'], ['test'])
- def test_get_project_larger_family(self):
- ''' Test the get_project_family from the internal API. '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- # Create a 3rd user
- item = pagure.lib.model.User(
- user='ralph',
- fullname='Ralph bar',
- password='ralph_foo',
- default_email='ralph@bar.com',
- )
- self.session.add(item)
- item = pagure.lib.model.UserEmail(
- user_id=3,
- email='ralph@bar.com')
- self.session.add(item)
- self.session.commit()
- # Create a couple of forks of the test project
- item = pagure.lib.model.Project(
- user_id=2, # foo
- name='test',
- is_fork=True,
- parent_id=1, # test
- description='test project #1',
- hook_token='aaabbbcccddd',
- )
- item.close_status = ['Invalid', 'Insufficient data', 'Fixed', 'Duplicate']
- self.session.add(item)
- item = pagure.lib.model.Project(
- user_id=3, # Ralph
- name='test',
- is_fork=True,
- parent_id=1, # test
- description='test project #1',
- hook_token='aaabbbccceee',
- )
- item.close_status = ['Invalid', 'Insufficient data', 'Fixed', 'Duplicate']
- self.session.add(item)
- self.session.commit()
- # Get on with testing
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- data = {
- 'csrf_token': csrf_token,
- }
- output = self.app.post('/pv/test/family', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'family']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(
- js_data['family'],
- ['test', 'fork/foo/test', 'fork/ralph/test'])
- def test_get_pull_request_ready_branch_main_repo_no_branch(self):
- '''Test the get_pull_request_ready_branch from the internal API
- on the main repository
- '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- # Get on with testing
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Query branches on the main repo
- data = {
- 'csrf_token': csrf_token,
- 'repo': 'test',
- }
- output = self.app.post('/pv/pull-request/ready', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(
- js_data['message'],
- {'branch_w_pr': {}, 'new_branch': {}})
- def test_get_pull_request_ready_branch_on_fork(self):
- '''Test the get_pull_request_ready_branch from the internal API on
- a fork
- '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.create_projects_git(
- os.path.join(self.path, 'repos', 'forks', 'foo'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'forks', 'foo', 'test.git'),
- branch='feature')
- # Create foo's fork of the test project
- item = pagure.lib.model.Project(
- user_id=2, # foo
- name='test',
- is_fork=True,
- parent_id=1, # test
- description='test project #1',
- hook_token='aaabbbcccddd',
- )
- item.close_status = ['Invalid', 'Insufficient data', 'Fixed', 'Duplicate']
- self.session.add(item)
- self.session.commit()
- # Get on with testing
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Query branches on the Ralph's fork
- data = {
- 'csrf_token': csrf_token,
- 'repo': 'test',
- 'repouser': 'foo',
- }
- output = self.app.post('/pv/pull-request/ready', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertListEqual(
- sorted(js_data['message'].keys()),
- ['branch_w_pr', 'new_branch'])
- self.assertEqual(js_data['message']['branch_w_pr'], {})
- self.assertListEqual(
- list(js_data['message']['new_branch']), ['feature'])
- self.assertEqual(len(js_data['message']['new_branch']['feature']), 2)
- def test_get_pull_request_ready_branch_on_fork_no_parent_no_pr(self):
- '''Test the get_pull_request_ready_branch from the internal API on
- a fork that has no parent repo (deleted) and doesn't allow PR
- '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.create_projects_git(
- os.path.join(self.path, 'repos', 'forks', 'foo'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'forks', 'foo', 'test.git'),
- branch='feature')
- # Create foo's fork of the test project
- item = pagure.lib.model.Project(
- user_id=2, # foo
- name='test',
- is_fork=True,
- parent_id=1, # test
- description='test project #1',
- hook_token='aaabbbcccddd',
- )
- item.close_status = ['Invalid', 'Insufficient data', 'Fixed', 'Duplicate']
- self.session.add(item)
- self.session.commit()
- settings = item.settings
- settings['pull_requests'] = False
- item.settings = settings
- self.session.add(item)
- self.session.commit()
- # Delete the parent project
- project = pagure.lib.get_authorized_project(self.session, 'test')
- self.session.delete(project)
- self.session.commit()
- # Get on with testing
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Query branches on the Ralph's fork
- data = {
- 'csrf_token': csrf_token,
- 'repo': 'test',
- 'repouser': 'foo',
- }
- output = self.app.post('/pv/pull-request/ready', data=data)
- self.assertEqual(output.status_code, 400)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message']
- )
- self.assertEqual(js_data['code'], 'ERROR')
- self.assertEqual(
- js_data['message'],
- 'Pull-request have been disabled for this repo')
- def test_get_pull_request_ready_branch_on_fork_no_parent(self):
- '''Test the get_pull_request_ready_branch from the internal API on
- a fork that has no parent repo (deleted).
- '''
- tests.create_projects(self.session)
- tests.create_projects_git(
- os.path.join(self.path, 'repos'), bare=True)
- tests.create_projects_git(
- os.path.join(self.path, 'repos', 'forks', 'foo'), bare=True)
- tests.add_content_git_repo(
- os.path.join(self.path, 'repos', 'forks', 'foo', 'test.git'),
- branch='feature')
- # Create foo's fork of the test project
- item = pagure.lib.model.Project(
- user_id=2, # foo
- name='test',
- is_fork=True,
- parent_id=1, # test
- description='test project #1',
- hook_token='aaabbbcccddd',
- )
- item.close_status = ['Invalid', 'Insufficient data', 'Fixed', 'Duplicate']
- self.session.add(item)
- self.session.commit()
- settings = item.settings
- settings['pull_requests'] = True
- item.settings = settings
- self.session.add(item)
- self.session.commit()
- # Delete the parent project
- project = pagure.lib.get_authorized_project(self.session, 'test')
- self.session.delete(project)
- self.session.commit()
- # Get on with testing
- user = tests.FakeUser()
- user.username = 'pingou'
- with tests.user_set(self.app.application, user):
- csrf_token = self.get_csrf()
- # Query branches on the Ralph's fork
- data = {
- 'csrf_token': csrf_token,
- 'repo': 'test',
- 'repouser': 'foo',
- }
- output = self.app.post('/pv/pull-request/ready', data=data)
- self.assertEqual(output.status_code, 200)
- js_data = json.loads(output.get_data(as_text=True))
- self.assertEqual(
- sorted(js_data.keys()),
- ['code', 'message']
- )
- self.assertEqual(js_data['code'], 'OK')
- self.assertEqual(
- sorted(js_data['message'].keys()),
- ['branch_w_pr', 'new_branch'])
- self.assertEqual(js_data['message']['branch_w_pr'], {})
- self.assertListEqual(
- list(js_data['message']['new_branch']), ['feature'])
- self.assertEqual(len(js_data['message']['new_branch']['feature']), 2)
- if __name__ == '__main__':
- unittest.main(verbosity=2)
|