diff --git a/.gitignore b/.gitignore
index 79ccbc1968f3fce96e8aef24d61c9902f69e8738..817301365b49aebaebcbbeb0ed4d75253d6a1e67 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,3 +10,4 @@ dist
.coverage
debexpo.egg-info/*
development-arno.ini
+cover
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index c6ca2415d7b5123bf117be1496190c79666a34c5..5c4db6bb163cb44c87bb3106a13d4977d5695f2b 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -6,7 +6,6 @@ before_script:
.test_template: &test
script:
- - mkdir -p /tmp/debexpo/pub
- python2.7 setup.py develop
- python2.7 setup.py compile_catalog
- python2.7 setup.py nosetests --with-coverage --cover-erase --cover-html
diff --git a/debexpo/controllers/index.py b/debexpo/controllers/index.py
index 62187da5695a3dbf30e3283aba8cf82ea8bf5157..140cb2c00819461b6ce8e2624315ca0011594779 100644
--- a/debexpo/controllers/index.py
+++ b/debexpo/controllers/index.py
@@ -79,15 +79,6 @@ class IndexController(BaseController):
def intro_maintainers(self):
"""Return an introduction page for package maintainers"""
-
- # The template will need to look at the user details.
- if 'user_id' in session:
- log.debug('Getting user object for user_id = "%s"' % session['user_id'])
- self.user = meta.session.query(User).get(session['user_id'])
- c.user = self.user
- c.logged_in = True
- else:
- c.logged_in = False
-
+ c.config = config
return render('/index/intro-maintainers.mako')
diff --git a/debexpo/controllers/register.py b/debexpo/controllers/register.py
index e6ffcc62b9abf2f245c3937d6802492f268fa23e..8e6918fbccf693203738f62d25e0ecb1f4891638 100644
--- a/debexpo/controllers/register.py
+++ b/debexpo/controllers/register.py
@@ -123,10 +123,6 @@ class RegisterController(BaseController):
"""
log.debug('Activation request with key = %s' % id)
- if id is None:
- log.error('Key is None')
- abort(404, 'Key is None')
-
user = meta.session.query(User).filter_by(verification=id).first()
if user is not None:
diff --git a/debexpo/model/package_info.py b/debexpo/model/package_info.py
index ff1706e6a0fd4d155c47f28bc2437d240eb65f30..e9b2c838a05b09d4b97c31bd30c553f2ca9a1214 100644
--- a/debexpo/model/package_info.py
+++ b/debexpo/model/package_info.py
@@ -105,7 +105,7 @@ class PackageInfo(OrmObject):
break
else:
# No template file found, something weird happened
- return "%s (!! no template found)" % self.data
+ return "%s (!! no template found)" % self.data # pragma: no cover
return template.render_unicode(o = self, h = debexpo.lib.helpers)
diff --git a/debexpo/tests/functional/test_index.py b/debexpo/tests/functional/test_index.py
index 6b4a7a58cbf34a73c3b9085b626d9b3798b346ab..03ebf7116f11a742a768a89c5676dafb7560d94e 100644
--- a/debexpo/tests/functional/test_index.py
+++ b/debexpo/tests/functional/test_index.py
@@ -8,19 +8,6 @@ import pylons.test
class TestIndexController(TestController):
- def setUp(self):
- self.tempdir = mkdtemp()
-
- def tearDown(self):
- rmtree(self.tempdir)
-
- def _generate_temppage(self, filename, text):
- temppage = os.path.join(self.tempdir, filename)
- f = open(temppage, 'w')
- f.write(text)
- f.close()
- return temppage
-
def test_index(self):
# test a normal index page
testurl = url(controller='index', action='index')
@@ -32,11 +19,26 @@ class TestIndexController(TestController):
def test_contact(self):
response = self.app.get(url(controller='index', action='contact'))
+ testtext = 'Site contact
'
self.assertEquals(response.status_int, 200)
+ self.assertTrue(testtext in response)
+
+ def test_qa(self):
+ response = self.app.get(url(controller='index', action='qa'))
+ testtext = 'Questions & Answers
'
+ self.assertEquals(response.status_int, 200)
+ self.assertTrue(testtext in response)
+
+ def test_intro_reviewer(self):
+ response = self.app.get(url(controller='index',
+ action='intro-reviewers'))
+ testtext = 'Package reviews
'
+ self.assertEquals(response.status_int, 200)
+ self.assertTrue(testtext in response)
def test_intro_maintainers(self):
- testurl = url('intro-maintainers')
- response = self.app.get(testurl)
+ response = self.app.get(url(controller='index',
+ action='intro-maintainers'))
testtext = "{}".format('Introduction for maintainers: How will my',
' package get into Debian
')
self.assertEquals(response.status_int, 200)
diff --git a/debexpo/tests/functional/test_package.py b/debexpo/tests/functional/test_package.py
index 1febfb7d6c5dc71f3d194adeb1008b0a08efbd11..ddefddfdc09bea515ced446b49416b6f66396aaf 100644
--- a/debexpo/tests/functional/test_package.py
+++ b/debexpo/tests/functional/test_package.py
@@ -5,10 +5,14 @@ from debexpo.model.users import User
from debexpo.model.packages import Package
from debexpo.model.package_comments import PackageComment
from debexpo.model.package_versions import PackageVersion
+from debexpo.model.package_info import PackageInfo
from debexpo.model.source_packages import SourcePackage
from debexpo.model.package_subscriptions import PackageSubscription
+from debexpo.controllers.package import PackageController
from datetime import datetime
import pylons.test
+import md5
+
class TestPackageController(TestController):
@@ -16,69 +20,107 @@ class TestPackageController(TestController):
self._setup_models()
self._setup_example_user()
user = meta.session.query(User).filter(
- User.email=='email@example.com').one()
+ User.email == 'email@example.com').one()
package = Package(name='testpackage', user=user,
description='a test package')
meta.session.add(package)
- package_version = PackageVersion(package=package, version='1.0-1',
- maintainer='Test User ',
- section='Admin',
- distribution='unstable',
- qa_status=0,
- component='main',
- priority='optional',
- closes='',
- uploaded=datetime.now())
+ package_version = PackageVersion(
+ package=package,
+ version='1.0-1',
+ maintainer='Test User ',
+ section='Admin',
+ distribution='unstable',
+ qa_status=0,
+ component='main',
+ priority='optional',
+ closes='',
+ uploaded=datetime.now())
meta.session.add(package_version)
meta.session.add(SourcePackage(package_version=package_version))
meta.session.commit()
def tearDown(self):
package = meta.session.query(Package).filter(
- Package.name=='testpackage').first()
+ Package.name == 'testpackage').first()
if package:
package_versions = meta.session.query(PackageVersion).filter(
- PackageVersion.package==package).all()
+ PackageVersion.package == package).all()
for vers in package_versions:
meta.session.query(SourcePackage).filter(
- SourcePackage.package_version==vers).delete()
+ SourcePackage.package_version == vers).delete()
meta.session.delete(vers)
meta.session.delete(package)
meta.session.commit()
self._remove_example_user()
+ def _test_no_auth(self, action, redirect_login=True):
+ user = meta.session.query(User).filter(
+ User.email == 'email@example.com').one()
+ response = self.app.get(url(
+ controller='package', action=action, packagename='testpackage',
+ key=user.get_upload_key()))
+ self.assertEquals(response.status_int, 302)
+ if (redirect_login):
+ self.assertTrue(response.location.endswith(url('login')))
+
+ def _test_wrong_key(self, action):
+ self.app.post(url('login'), self._AUTHDATA)
+ user = meta.session.query(User).filter(
+ User.email == 'email@example.com').one()
+ response = self.app.get(url(
+ controller='package', action=action, packagename='testpackage',
+ key='wrong'), expect_errors=True)
+ self.assertEquals(response.status_int, 402)
+
+ def _test_not_owned_package(self, action):
+ user = User(name='Another user', email='another@example.com',
+ password=md5.new('password').hexdigest(),
+ lastlogin=datetime.now())
+ meta.session.add(user)
+ meta.session.commit()
+ self.app.post(url('login'), {'email': 'another@example.com',
+ 'password': 'password',
+ 'commit': 'submit'})
+ user = meta.session.query(User).filter(
+ User.email == 'another@example.com').one()
+ response = self.app.get(url(
+ controller='package', action=action, packagename='testpackage',
+ key=user.get_upload_key()), expect_errors=True)
+ self.assertEquals(response.status_int, 403)
+ meta.session.delete(user)
+
def test_index(self):
user = meta.session.query(User).filter(
- User.email=='email@example.com').one()
+ User.email == 'email@example.com').one()
response = self.app.get(url(controller='package', action='index'))
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url(controller='packages', action='index', packagename=None)))
+ url(controller='packages', action='index', packagename=None)))
response = self.app.get(url(controller='package', action='index',
packagename='notapackage'))
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url(controller='packages', action='index', packagename=None)))
+ url(controller='packages', action='index', packagename=None)))
response = self.app.get(url(controller='package', action='index',
packagename='testpackage'))
self.assertEquals(response.status_int, 200)
self.assertEquals(len(response.lxml.xpath(
- '//a[@href="%s"]' % url(controller='packages',
- action='uploader',
- id='email@example.com'))), 1)
+ '//a[@href="%s"]' % url(controller='packages',
+ action='uploader',
+ id='email@example.com'))), 1)
response = self.app.post(url('login'), self._AUTHDATA)
response = self.app.get(url(controller='package', action='index',
packagename='testpackage'))
self.assertEquals(response.status_int, 200)
self.assertEquals(len(response.lxml.xpath(
- '//a[@href="%s"]' % url(
- controller='package', action='delete',
- packagename='testpackage',
- key=user.get_upload_key()))), 1)
+ '//a[@href="%s"]' % url(
+ controller='package', action='delete',
+ packagename='testpackage',
+ key=user.get_upload_key()))), 1)
self.assertEquals(len(response.lxml.xpath(
- '//form[@action="%s"]' % url(
- controller='package', action='comment',
- packagename='testpackage'))), 1)
+ '//form[@action="%s"]' % url(
+ controller='package', action='comment',
+ packagename='testpackage'))), 1)
def test_subscribe(self):
response = self.app.get(url(controller='package', action='subscribe',
@@ -86,8 +128,8 @@ class TestPackageController(TestController):
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(url('login')))
self.assertEquals(response.session['path_before_login'], url(
- controller='package', action='subscribe',
- packagename='testpackage'))
+ controller='package', action='subscribe',
+ packagename='testpackage'))
self.app.post(url('login'), self._AUTHDATA)
response = self.app.get(url(controller='package', action='subscribe',
packagename='testpackage'))
@@ -101,22 +143,23 @@ class TestPackageController(TestController):
self.assertEquals(len(options), 3)
for option in options:
self.assertTrue(option.attrib['value'] in [
- str(item) for item in (
- -1, constants.SUBSCRIPTION_LEVEL_UPLOADS,
- constants.SUBSCRIPTION_LEVEL_COMMENTS)])
+ str(item) for item in (
+ -1, constants.SUBSCRIPTION_LEVEL_UPLOADS,
+ constants.SUBSCRIPTION_LEVEL_COMMENTS)])
response = self.app.post(url(controller='package', action='subscribe',
packagename='testpackage'))
self.assertEquals(response.status_int, 200)
self.assertEquals(
len(response.lxml.xpath(
- '//span[@class="error-message"]')), 2)
- response = self.app.post(url(controller='package', action='subscribe',
+ '//span[@class="error-message"]')), 2)
+ response = self.app.post(url(controller='package',
+ action='subscribe',
packagename='testpackage'),
{'level': constants.SUBSCRIPTION_LEVEL_UPLOADS,
'commit': 'submit'})
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url('package', packagename='testpackage')))
+ url('package', packagename='testpackage')))
subs = meta.session.query(PackageSubscription).filter_by(
package='testpackage').filter_by(
user_id=response.session['user_id']).one()
@@ -125,15 +168,16 @@ class TestPackageController(TestController):
packagename='testpackage'))
self.assertEquals(
len(response.lxml.xpath(
- '//option[@value="%d" and @selected="selected"]' %
- constants.SUBSCRIPTION_LEVEL_UPLOADS)), 1)
- response = self.app.post(url(controller='package', action='subscribe',
+ '//option[@value="%d" and @selected="selected"]' %
+ constants.SUBSCRIPTION_LEVEL_UPLOADS)), 1)
+ response = self.app.post(url(controller='package',
+ action='subscribe',
packagename='testpackage'),
{'level': constants.SUBSCRIPTION_LEVEL_COMMENTS,
'commit': 'submit'})
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url('package', packagename='testpackage')))
+ url('package', packagename='testpackage')))
subs = meta.session.query(PackageSubscription).filter_by(
package='testpackage').filter_by(
user_id=response.session['user_id']).one()
@@ -144,32 +188,57 @@ class TestPackageController(TestController):
'commit': 'submit'})
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url('package', packagename='testpackage')))
+ url('package', packagename='testpackage')))
subs = meta.session.query(PackageSubscription).filter_by(
package='testpackage').filter_by(
user_id=response.session['user_id']).first()
self.assertEquals(subs, None)
- def test_delete(self):
- user = meta.session.query(User).filter(
- User.email=='email@example.com').one()
- response = self.app.get(url(
- controller='package', action='delete', packagename='testpackage',
- key=user.get_upload_key()))
- self.assertEquals(response.status_int, 302)
- self.assertTrue(response.location.endswith(url('login')))
+ def test_get_package_from_crontab_wrong_package(self):
+ pkg_controller = PackageController()
+ package = pkg_controller._get_package('nonexistant',
+ from_controller=False)
+ self.assertEquals(package, None)
+
+ def test_get_package_from_crontab(self):
+ pkg_controller = PackageController()
+ package = pkg_controller._get_package('testpackage',
+ from_controller=False)
+ self.assertEquals(package.name, 'testpackage')
+
+ def test_delete_no_auth(self):
+ self._test_no_auth('delete')
+
+ def test_delete_not_owned_package(self):
+ self._test_not_owned_package('delete')
+
+ def test_delete_wrong_key(self):
+ self._test_wrong_key('delete')
+
+ def test_delete_successful(self):
self.app.post(url('login'), self._AUTHDATA)
+ user = meta.session.query(User).filter(
+ User.email == 'email@example.com').one()
response = self.app.get(url(
controller='package', action='delete', packagename='testpackage',
key=user.get_upload_key()))
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url(controller='packages', action='my')))
+ url(controller='packages', action='my')))
package = meta.session.query(Package).filter(
- Package.name=='testpackage').first()
+ Package.name == 'testpackage').first()
self.assertEquals(package, None)
- def test_comment(self):
+ def test_comment_show(self):
+ response = self.app.get(
+ url(controller='package', action='comment',
+ packagename='testpackage'))
+ self.assertEquals(response.status_int, 302)
+ self.assertTrue(response.location.endswith(
+ url(controller='package', action='index',
+ packagename='testpackage')))
+
+ def test_comment_no_auth(self):
response = self.app.post(
url(controller='package', action='comment',
packagename='testpackage'),
@@ -179,6 +248,8 @@ class TestPackageController(TestController):
'commit': 'submit'})
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(url('login')))
+
+ def test_comment(self):
self.app.post(url('login'), self._AUTHDATA)
response = self.app.post(
url(controller='package', action='comment',
@@ -189,17 +260,21 @@ class TestPackageController(TestController):
'commit': 'submit'})
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url(controller='package', action='index',
- packagename='testpackage')))
+ url(controller='package', action='index',
+ packagename='testpackage')))
comment = meta.session.query(PackageComment).filter_by(
package_version_id=1).one()
self.assertEquals(comment.text, 'This is a test comment')
self.assertEquals(comment.status,
constants.PACKAGE_COMMENT_STATUS_NOT_UPLOADED)
meta.session.delete(comment)
+
+ def test_comment_with_subscriber(self):
# test with a subscriber
pylons.test.pylonsapp.config['debexpo.testsmtp'] = '/tmp/debexpo.msg'
- user = meta.session.query(User).filter(User.email=='email@example.com').one()
+ self.app.post(url('login'), self._AUTHDATA)
+ user = meta.session.query(User).filter(
+ User.email == 'email@example.com').one()
packsub = PackageSubscription(
package='testpackage',
level=constants.SUBSCRIPTION_LEVEL_COMMENTS)
@@ -216,8 +291,8 @@ class TestPackageController(TestController):
'status': 'checked'})
self.assertEquals(response.status_int, 302)
self.assertTrue(response.location.endswith(
- url(controller='package', action='index',
- packagename='testpackage')))
+ url(controller='package', action='index',
+ packagename='testpackage')))
comment = meta.session.query(PackageComment).filter_by(
package_version_id=1).one()
self.assertEquals(comment.text, 'This is a test comment')
@@ -225,3 +300,66 @@ class TestPackageController(TestController):
constants.PACKAGE_COMMENT_STATUS_UPLOADED)
meta.session.delete(packsub)
meta.session.commit()
+
+ def test_sponsor_no_auth(self):
+ self._test_no_auth('sponsor')
+
+ def test_sponsor_wrong_key(self):
+ self._test_wrong_key('sponsor')
+
+ def test_sponsor_not_owned_package(self):
+ self._test_not_owned_package('sponsor')
+
+ def test_sponsor_toggle(self, toggle=True):
+ self.app.post(url('login'), self._AUTHDATA)
+ user = meta.session.query(User).filter(
+ User.email == 'email@example.com').one()
+ response = self.app.get(url(
+ controller='package', action='sponsor', packagename='testpackage',
+ key=user.get_upload_key()))
+ self.assertEquals(response.status_int, 302)
+ package = meta.session.query(Package).filter(
+ Package.name == 'testpackage').first()
+ if (toggle):
+ self.assertFalse(package.needs_sponsor)
+ self.test_sponsor_toggle(toggle=False)
+ else:
+ self.assertTrue(package.needs_sponsor)
+
+ def test_package_info_rich_data(self):
+ package = meta.session.query(Package).filter(
+ Package.name == 'testpackage').first()
+ package_version = meta.session.query(PackageVersion).filter(
+ PackageVersion.package == package).first()
+ textmark = 'some lintian output'
+ package_info_data = PackageInfo(
+ package_version_id=package_version.id,
+ from_plugin='LintianPlugin',
+ outcome='Package is lintian clean',
+ rich_data=textmark,
+ severity=constants.PLUGIN_SEVERITY_INFO)
+ meta.session.add(package_info_data)
+ meta.session.commit()
+ response = self.app.get(url(controller='package', action='index',
+ packagename='testpackage'))
+ self.assertEquals(response.status_int, 200)
+ self.assertTrue(textmark in response)
+
+ def test_package_info_data(self):
+ package = meta.session.query(Package).filter(
+ Package.name == 'testpackage').first()
+ package_version = meta.session.query(PackageVersion).filter(
+ PackageVersion.package == package).first()
+ textmark = 'some lintian output'
+ package_info_data = PackageInfo(
+ package_version_id=package_version.id,
+ from_plugin='LintianPlugin',
+ outcome='Package is lintian clean',
+ data=textmark,
+ severity=constants.PLUGIN_SEVERITY_INFO)
+ meta.session.add(package_info_data)
+ meta.session.commit()
+ response = self.app.get(url(controller='package', action='index',
+ packagename='testpackage'))
+ self.assertEquals(response.status_int, 200)
+ self.assertTrue(textmark in response)
diff --git a/debexpo/tests/functional/test_register.py b/debexpo/tests/functional/test_register.py
index f68f5508cbaa8fe8145cf540a1a4b6ced3d0a34d..2e928c878789fedee9a0b84eaddec187ee29afb4 100644
--- a/debexpo/tests/functional/test_register.py
+++ b/debexpo/tests/functional/test_register.py
@@ -1,6 +1,7 @@
from debexpo.tests import TestController, url
from debexpo.model import meta
from debexpo.model.users import User
+from debexpo.lib import constants
import pylons.test
class TestRegisterController(TestController):
@@ -42,7 +43,8 @@ class TestRegisterController(TestController):
'email': 'mr_me_again@example.com',
'sponsor': '0'})
- count = meta.session.query(User).filter(User.email=='mr_me_again@example.com').count()
+ count = meta.session.query(User).filter(
+ User.email=='mr_me_again@example.com').count()
self.assertEquals(count, 0)
# The assertion is that there are no matching users by
# email address.
@@ -58,6 +60,58 @@ class TestRegisterController(TestController):
# Now, finally, delete that User that we created.
meta.session.delete(
- meta.session.query(User).filter(User.email=='mr_me@example.com').one())
+ meta.session.query(User).filter(
+ User.email=='mr_me@example.com').one())
+ def test_sponsor_signup_wrong_email(self):
+ pylons.test.pylonsapp.config['debexpo.testsmtp'] = '/tmp/debexpo.msg'
+ response = self.app.post(url(controller='register', action='register'),
+ {'name': 'Mr. Me',
+ 'password': 'password',
+ 'password_confirm': 'password',
+ 'commit': 'yes',
+ 'email': 'mr_me@example.org',
+ 'sponsor': '1'})
+ self.assertEqual(response.status_int, 200)
+ testtext = '{}'.format('A sponsor account must be registered with your',
+ ' @debian.org address')
+ self.assertTrue(testtext in response)
+ count = meta.session.query(User).filter(User.email=='mr_me@example.com').count()
+ self.assertEquals(count, 0)
+
+
+ def test_sponsor_signup(self):
+ pylons.test.pylonsapp.config['debexpo.testsmtp'] = '/tmp/debexpo.msg'
+ response = self.app.post(url(controller='register', action='register'),
+ {'name': 'Mr. Me Debian',
+ 'password': 'password',
+ 'password_confirm': 'password',
+ 'commit': 'yes',
+ 'email': 'mr_me@debian.org',
+ 'sponsor': '1'})
+
+ self.assertEqual(response.status_int, 200)
+ user = meta.session.query(User).filter(
+ User.email=='mr_me@debian.org').first()
+ self.assertEquals(user.status, constants.USER_STATUS_DEVELOPER)
+
+ user = meta.session.query(User).filter(
+ User.email=='mr_me@debian.org').one()
+ meta.session.delete(user)
+
+ def test_activation_wrong_key(self):
+ response = self.app.get(url(controller='register', action='activate',
+ id='that_key_should_not_exist'),
+ expect_errors=True)
+ self.assertEqual(response.status_int, 404)
+
+ def test_successful_activation(self):
+ self.test_maintainer_signup(actually_delete_it=False)
+ user = meta.session.query(User).filter(
+ User.email=='mr_me@example.com').one()
+ response = self.app.get(url(controller='register', action='activate',
+ id=user.verification),
+ expect_errors=True)
+ self.assertEqual(response.status_int, 200)
+ meta.session.delete(user)
diff --git a/debexpo/tests/functional/test_upload.py b/debexpo/tests/functional/test_upload.py
index 52b29d29d1b9b0c0e73835a9b1c91ea0a6987dc8..5d6d61fb1ad9ba3479fb9e5d7f14704373dddb38 100644
--- a/debexpo/tests/functional/test_upload.py
+++ b/debexpo/tests/functional/test_upload.py
@@ -42,6 +42,7 @@ import base64
from debexpo.tests import TestController, url
import pylons.test
+
class TestUploadController(TestController):
def __init__(self, *args, **kwargs):
@@ -59,16 +60,20 @@ class TestUploadController(TestController):
def testGetRequest(self):
"""
- Tests whether requests where method != PUT are rejected with error code 405.
+ Tests whether requests where method != PUT are rejected with error code
+ 405.
"""
- response = self.app.get(url(controller='upload', action='index',
- filename='testname.dsc'), expect_errors=True)
+ response = self.app.get(url(controller='upload',
+ action='index',
+ filename='testname.dsc'),
+ expect_errors=True)
self.assertEqual(response.status_int, 405)
def testExtensionNotAllowed(self):
"""
- Tests whether uploads of an unknown file extensions are rejected with error code 403.
+ Tests whether uploads of an unknown file extensions are rejected with
+ error code 403.
"""
response = self.app.put(url(controller='upload', action='index',
filename='testname.unknown'),
@@ -78,23 +83,31 @@ class TestUploadController(TestController):
def testSuccessfulUpload(self):
"""
- Tests whether uploads with sane file extensions and authorization are successful.
+ Tests whether uploads with sane file extensions and authorization are
+ successful.
"""
response = self.app.put(url(
- controller='upload', action='index',
- filename='testfile2.dsc'),
+ controller='upload', action='index',
+ filename='testfile2.dsc'),
params='contents', expect_errors=False)
self.assertEqual(response.status_int, 200)
app_config = pylons.test.pylonsapp.config
- self.assertTrue(os.path.isfile(os.path.join(app_config['debexpo.upload.incoming'],
- 'pub',
- 'testfile2.dsc')))
-
- self.assertEqual(file(os.path.join(app_config['debexpo.upload.incoming'],
- 'pub',
- 'testfile2.dsc')).read(), 'contents')
+ self.assertTrue(
+ os.path.isfile(
+ os.path.join(
+ app_config['debexpo.upload.incoming'],
+ 'pub',
+ 'testfile2.dsc')))
+
+ self.assertEqual(
+ file(
+ os.path.join(
+ app_config['debexpo.upload.incoming'],
+ 'pub',
+ 'testfile2.dsc')).read(),
+ 'contents')
if os.path.isfile(os.path.join(app_config['debexpo.upload.incoming'],
'pub', 'testfile2.dsc')):
@@ -106,15 +119,15 @@ class TestUploadController(TestController):
Tests whether a re-uploads of the same file failed with error code 403.
"""
response = self.app.put(url(
- controller='upload', action='index',
- filename='testfile.dsc'),
+ controller='upload', action='index',
+ filename='testfile.dsc'),
params='contents', expect_errors=False)
self.assertEqual(response.status_int, 200)
response = self.app.put(url(
- controller='upload', action='index',
- filename='testfile.dsc'),
+ controller='upload', action='index',
+ filename='testfile.dsc'),
params='contents', expect_errors=True)
self.assertEqual(response.status_int, 403)
@@ -125,3 +138,44 @@ class TestUploadController(TestController):
'pub', 'testfile.dsc')):
os.remove(os.path.join(app_config['debexpo.upload.incoming'],
'pub', 'testfile.dsc'))
+
+ def testUploadWithoutConfig(self):
+ """
+ Tests whether an uploads without debexpo.upload.incoming fails.
+ """
+ pylons.test.pylonsapp.config.pop('debexpo.upload.incoming')
+
+ response = self.app.put(url(
+ controller='upload', action='index',
+ filename='testfile.dsc'),
+ params='contents', expect_errors=True)
+
+ self.assertEqual(response.status_int, 500)
+
+ def testUploadNonexistantQueue(self):
+ """
+ Tests whether an uploads with an nonexistant queue fails.
+ """
+ app_config = pylons.test.pylonsapp.config
+ app_config['debexpo.upload.incoming'] = '/nonexistant'
+
+ response = self.app.put(url(
+ controller='upload', action='index',
+ filename='testfile.dsc'),
+ params='contents', expect_errors=True)
+
+ self.assertEqual(response.status_int, 500)
+
+ def testUploadNonwritableQueue(self):
+ """
+ Tests whether an uploads with an nonwritable queue fails.
+ """
+ app_config = pylons.test.pylonsapp.config
+ app_config['debexpo.upload.incoming'] = '/proc/sys'
+
+ response = self.app.put(url(
+ controller='upload', action='index',
+ filename='testfile.dsc'),
+ params='contents', expect_errors=True)
+
+ self.assertEqual(response.status_int, 500)
diff --git a/debexpo/websetup.py b/debexpo/websetup.py
index cfda9f623793ad1518d44d8da17d0c2d32b4338d..a322f2eca8893232eb3cc53178c3ae25aca6e0af 100644
--- a/debexpo/websetup.py
+++ b/debexpo/websetup.py
@@ -84,6 +84,7 @@ def setup_config(command, filename, section, vars):
if not os.path.isdir(config['debexpo.upload.incoming']):
log.info('Creating incoming directory')
os.mkdir(config['debexpo.upload.incoming'])
+ os.mkdir(os.path.join(config['debexpo.upload.incoming'], 'pub'))
else:
log.info('Incoming directory already exists')