launchpad-reviewers team mailing list archive
-
launchpad-reviewers team
-
Mailing list archive
-
Message #24122
[Merge] ~cjwatson/launchpad:faster-blueprints-webservice-tests into launchpad:master
Colin Watson has proposed merging ~cjwatson/launchpad:faster-blueprints-webservice-tests into launchpad:master.
Commit message:
Stop using launchpadlib in lp.blueprints.tests.test_webservice
Requested reviews:
Launchpad code reviewers (launchpad-reviewers)
For more details, see:
https://code.launchpad.net/~cjwatson/launchpad/+git/launchpad/+merge/375664
Port the blueprints webservice tests to use in-process webservice calls rather than launchpadlib and AppServerLayer. While the code is a bit longer as a result, it's easier to debug and substantially faster: this change takes the test time for this file from 149 seconds to 63 seconds on my laptop. It also removes the last remaining dependency on AppServerLayer's SMTP server outside of the Mailman integration tests.
--
Your team Launchpad code reviewers is requested to review the proposed merge of ~cjwatson/launchpad:faster-blueprints-webservice-tests into launchpad:master.
diff --git a/lib/lp/blueprints/tests/test_webservice.py b/lib/lp/blueprints/tests/test_webservice.py
index c1bf31f..3235b05 100644
--- a/lib/lp/blueprints/tests/test_webservice.py
+++ b/lib/lp/blueprints/tests/test_webservice.py
@@ -1,4 +1,4 @@
-# Copyright 2009 Canonical Ltd. This software is licensed under the
+# Copyright 2009-2019 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Webservice unit tests related to Launchpad blueprints."""
@@ -9,12 +9,17 @@ __metaclass__ = type
import json
-from testtools.matchers import MatchesStructure
-import transaction
-from zope.security.management import endInteraction
-from zope.security.proxy import removeSecurityProxy
+import iso8601
+from testtools.matchers import (
+ AfterPreprocessing,
+ ContainsDict,
+ Equals,
+ MatchesListwise,
+ )
+from zope.component import getUtility
from lp.app.enums import InformationType
+from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.blueprints.enums import SpecificationDefinitionStatus
from lp.registry.enums import SpecificationSharingPolicy
from lp.services.webapp.interaction import ANONYMOUS
@@ -22,45 +27,21 @@ from lp.services.webapp.interfaces import OAuthPermission
from lp.testing import (
admin_logged_in,
api_url,
- launchpadlib_for,
+ login,
+ logout,
person_logged_in,
TestCaseWithFactory,
- ws_object,
- )
-from lp.testing.layers import (
- AppServerLayer,
- DatabaseFunctionalLayer,
)
+from lp.testing.layers import DatabaseFunctionalLayer
from lp.testing.pages import (
LaunchpadWebServiceCaller,
webservice_for_person,
)
-class SpecificationWebserviceTestCase(TestCaseWithFactory):
-
- def getLaunchpadlib(self):
- user = self.factory.makePerson()
- return launchpadlib_for("testing", user, version='devel')
-
- def getSpecOnWebservice(self, spec_object):
- launchpadlib = self.getLaunchpadlib()
- # Ensure that there is an interaction so that the security
- # checks for spec_object work.
- with person_logged_in(ANONYMOUS):
- url = '/%s/+spec/%s' % (spec_object.target.name, spec_object.name)
- result = launchpadlib.load(url)
- return result
-
- def getPillarOnWebservice(self, pillar_obj):
- pillar_name = pillar_obj.name
- launchpadlib = self.getLaunchpadlib()
- return launchpadlib.load(pillar_name)
-
-
-class SpecificationWebserviceTests(SpecificationWebserviceTestCase):
+class SpecificationWebserviceTests(TestCaseWithFactory):
"""Test accessing specification top-level webservice."""
- layer = AppServerLayer
+ layer = DatabaseFunctionalLayer
def test_collection(self):
# `ISpecificationSet` is exposed as a webservice via /specs
@@ -132,9 +113,9 @@ class SpecificationWebserviceTests(SpecificationWebserviceTestCase):
self.assertEqual(201, response.status)
-class SpecificationAttributeWebserviceTests(SpecificationWebserviceTestCase):
+class SpecificationAttributeWebserviceTests(TestCaseWithFactory):
"""Test accessing specification attributes over the webservice."""
- layer = AppServerLayer
+ layer = DatabaseFunctionalLayer
def test_representation_is_empty_on_1_dot_0(self):
# ISpecification is exposed on the 1.0 version so that they can be
@@ -153,81 +134,138 @@ class SpecificationAttributeWebserviceTests(SpecificationWebserviceTestCase):
def test_representation_basics(self):
spec = self.factory.makeSpecification()
- spec_webservice = self.getSpecOnWebservice(spec)
+ spec_url = api_url(spec)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
with person_logged_in(ANONYMOUS):
self.assertThat(
- spec_webservice,
- MatchesStructure.byEquality(
- name=spec.name,
- title=spec.title,
- specification_url=spec.specurl,
- summary=spec.summary,
- implementation_status=spec.implementation_status.title,
- definition_status=spec.definition_status.title,
- priority=spec.priority.title,
- date_created=spec.datecreated,
- whiteboard=spec.whiteboard,
- workitems_text=spec.workitems_text))
+ response.jsonBody(),
+ ContainsDict({
+ 'name': Equals(spec.name),
+ 'title': Equals(spec.title),
+ 'specification_url': Equals(spec.specurl),
+ 'summary': Equals(spec.summary),
+ 'implementation_status': Equals(
+ spec.implementation_status.title),
+ 'definition_status': Equals(
+ spec.definition_status.title),
+ 'priority': Equals(spec.priority.title),
+ 'date_created': AfterPreprocessing(
+ iso8601.parse_date, Equals(spec.datecreated)),
+ 'whiteboard': Equals(spec.whiteboard),
+ 'workitems_text': Equals(spec.workitems_text),
+ }))
def test_representation_contains_target(self):
spec = self.factory.makeSpecification(
product=self.factory.makeProduct())
- spec_target_name = spec.target.name
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual(spec_target_name, spec_webservice.target.name)
+ spec_url = api_url(spec)
+ spec_target_url = api_url(spec.target)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ self.assertEndsWith(
+ response.jsonBody()['target_link'], spec_target_url)
def test_representation_contains_assignee(self):
- # Hard-code the person's name or else we'd need to set up a zope
- # interaction as IPerson.name is protected.
spec = self.factory.makeSpecification(
- assignee=self.factory.makePerson(name='test-person'))
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual('test-person', spec_webservice.assignee.name)
+ assignee=self.factory.makePerson())
+ spec_url = api_url(spec)
+ spec_assignee_url = api_url(spec.assignee)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ self.assertEndsWith(
+ response.jsonBody()['assignee_link'], spec_assignee_url)
def test_representation_contains_drafter(self):
spec = self.factory.makeSpecification(
- drafter=self.factory.makePerson(name='test-person'))
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual('test-person', spec_webservice.drafter.name)
+ drafter=self.factory.makePerson())
+ spec_url = api_url(spec)
+ spec_drafter_url = api_url(spec.drafter)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ self.assertEndsWith(
+ response.jsonBody()['drafter_link'], spec_drafter_url)
def test_representation_contains_approver(self):
spec = self.factory.makeSpecification(
- approver=self.factory.makePerson(name='test-person'))
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual('test-person', spec_webservice.approver.name)
+ approver=self.factory.makePerson())
+ spec_url = api_url(spec)
+ spec_approver_url = api_url(spec.approver)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ self.assertEndsWith(
+ response.jsonBody()['approver_link'], spec_approver_url)
def test_representation_contains_owner(self):
- spec = self.factory.makeSpecification(
- owner=self.factory.makePerson(name='test-person'))
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual('test-person', spec_webservice.owner.name)
+ spec = self.factory.makeSpecification(owner=self.factory.makePerson())
+ spec_url = api_url(spec)
+ spec_owner_url = api_url(spec.owner)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ self.assertEndsWith(response.jsonBody()['owner_link'], spec_owner_url)
def test_representation_contains_milestone(self):
product = self.factory.makeProduct()
productseries = self.factory.makeProductSeries(product=product)
milestone = self.factory.makeMilestone(
- name="1.0", product=product, productseries=productseries)
+ product=product, productseries=productseries)
+ milestone_url = api_url(milestone)
spec_object = self.factory.makeSpecification(
product=product, goal=productseries, milestone=milestone)
- spec = self.getSpecOnWebservice(spec_object)
- self.assertEqual("1.0", spec.milestone.name)
+ spec_object_url = api_url(spec_object)
+ webservice = webservice_for_person(
+ spec_object.owner, default_api_version='devel')
+ response = webservice.get(spec_object_url)
+ self.assertEqual(200, response.status)
+ self.assertEndsWith(
+ response.jsonBody()['milestone_link'], milestone_url)
def test_representation_contains_dependencies(self):
spec = self.factory.makeSpecification()
spec2 = self.factory.makeSpecification()
spec2_name = spec2.name
spec.createDependency(spec2)
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual(1, spec_webservice.dependencies.total_size)
- self.assertEqual(spec2_name, spec_webservice.dependencies[0].name)
+ spec_url = api_url(spec)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ response = webservice.get(
+ response.jsonBody()['dependencies_collection_link'])
+ self.assertEqual(200, response.status)
+ self.assertThat(response.jsonBody(), ContainsDict({
+ 'total_size': Equals(1),
+ 'entries': MatchesListwise([
+ ContainsDict({'name': Equals(spec2_name)}),
+ ]),
+ }))
def test_representation_contains_linked_branches(self):
spec = self.factory.makeSpecification()
branch = self.factory.makeBranch()
person = self.factory.makePerson()
spec.linkBranch(branch, person)
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual(1, spec_webservice.linked_branches.total_size)
+ spec_url = api_url(spec)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ response = webservice.get(
+ response.jsonBody()['linked_branches_collection_link'])
+ self.assertEqual(200, response.status)
+ self.assertEqual(1, response.jsonBody()['total_size'])
def test_representation_contains_bug_links(self):
spec = self.factory.makeSpecification()
@@ -235,9 +273,17 @@ class SpecificationAttributeWebserviceTests(SpecificationWebserviceTestCase):
person = self.factory.makePerson()
with person_logged_in(person):
spec.linkBug(bug)
- spec_webservice = self.getSpecOnWebservice(spec)
- self.assertEqual(1, spec_webservice.bugs.total_size)
- self.assertEqual(bug.id, spec_webservice.bugs[0].id)
+ spec_url = api_url(spec)
+ webservice = webservice_for_person(
+ spec.owner, default_api_version='devel')
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ response = webservice.get(
+ response.jsonBody()['bugs_collection_link'])
+ self.assertThat(response.jsonBody(), ContainsDict({
+ 'total_size': Equals(1),
+ 'entries': MatchesListwise([ContainsDict({'id': Equals(bug.id)})]),
+ }))
class SpecificationMutationTests(TestCaseWithFactory):
@@ -289,89 +335,121 @@ class SpecificationMutationTests(TestCaseWithFactory):
"There is already a blueprint named foo for Fooix.", response.body)
-class SpecificationTargetTests(SpecificationWebserviceTestCase):
+class SpecificationTargetTests(TestCaseWithFactory):
"""Tests for accessing specifications via their targets."""
- layer = AppServerLayer
+ layer = DatabaseFunctionalLayer
def test_get_specification_on_product(self):
product = self.factory.makeProduct(name="fooix")
self.factory.makeSpecification(
product=product, name="some-spec")
- product_on_webservice = self.getPillarOnWebservice(product)
- spec = product_on_webservice.getSpecification(name="some-spec")
- self.assertEqual("some-spec", spec.name)
- self.assertEqual("fooix", spec.target.name)
+ product_url = api_url(product)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.named_get(
+ product_url, "getSpecification", name="some-spec")
+ self.assertEqual(200, response.status)
+ self.assertEqual("some-spec", response.jsonBody()["name"])
+ response = webservice.get(response.jsonBody()["target_link"])
+ self.assertEqual(200, response.status)
+ self.assertEqual("fooix", response.jsonBody()["name"])
def test_get_specification_on_distribution(self):
distribution = self.factory.makeDistribution(name="foobuntu")
self.factory.makeSpecification(
distribution=distribution, name="some-spec")
- distro_on_webservice = self.getPillarOnWebservice(distribution)
- spec = distro_on_webservice.getSpecification(name="some-spec")
- self.assertEqual("some-spec", spec.name)
- self.assertEqual("foobuntu", spec.target.name)
+ distribution_url = api_url(distribution)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.named_get(
+ distribution_url, "getSpecification", name="some-spec")
+ self.assertEqual(200, response.status)
+ self.assertEqual("some-spec", response.jsonBody()["name"])
+ response = webservice.get(response.jsonBody()["target_link"])
+ self.assertEqual(200, response.status)
+ self.assertEqual("foobuntu", response.jsonBody()["name"])
def test_get_specification_on_productseries(self):
product = self.factory.makeProduct(name="fooix")
- productseries = self.factory.makeProductSeries(
- product=product, name="fooix-dev")
+ productseries = self.factory.makeProductSeries(product=product)
self.factory.makeSpecification(
product=product, name="some-spec", goal=productseries)
- product_on_webservice = self.getPillarOnWebservice(product)
- productseries_on_webservice = product_on_webservice.getSeries(
- name="fooix-dev")
- spec = productseries_on_webservice.getSpecification(name="some-spec")
- self.assertEqual("some-spec", spec.name)
- self.assertEqual("fooix", spec.target.name)
+ productseries_url = api_url(productseries)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.named_get(
+ productseries_url, "getSpecification", name="some-spec")
+ self.assertEqual(200, response.status)
+ self.assertEqual("some-spec", response.jsonBody()["name"])
+ response = webservice.get(response.jsonBody()["target_link"])
+ self.assertEqual(200, response.status)
+ self.assertEqual("fooix", response.jsonBody()["name"])
def test_get_specification_on_distroseries(self):
distribution = self.factory.makeDistribution(name="foobuntu")
distroseries = self.factory.makeDistroSeries(
- distribution=distribution, name="maudlin")
+ distribution=distribution)
self.factory.makeSpecification(
distribution=distribution, name="some-spec",
goal=distroseries)
- distro_on_webservice = self.getPillarOnWebservice(distribution)
- distroseries_on_webservice = distro_on_webservice.getSeries(
- name_or_version="maudlin")
- spec = distroseries_on_webservice.getSpecification(name="some-spec")
- self.assertEqual("some-spec", spec.name)
- self.assertEqual("foobuntu", spec.target.name)
+ distroseries_url = api_url(distroseries)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.named_get(
+ distroseries_url, "getSpecification", name="some-spec")
+ self.assertEqual(200, response.status)
+ self.assertEqual("some-spec", response.jsonBody()["name"])
+ response = webservice.get(response.jsonBody()["target_link"])
+ self.assertEqual(200, response.status)
+ self.assertEqual("foobuntu", response.jsonBody()["name"])
def test_get_specification_not_found(self):
product = self.factory.makeProduct()
- product_on_webservice = self.getPillarOnWebservice(product)
- spec = product_on_webservice.getSpecification(name="nonexistant")
- self.assertEqual(None, spec)
+ product_url = api_url(product)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.named_get(
+ product_url, "getSpecification", name="nonexistent")
+ self.assertEqual(200, response.status)
+ self.assertIsNone(response.jsonBody())
-class IHasSpecificationsTests(SpecificationWebserviceTestCase):
+class IHasSpecificationsTests(TestCaseWithFactory):
"""Tests for accessing IHasSpecifications methods over the webservice."""
layer = DatabaseFunctionalLayer
- def assertNamesOfSpecificationsAre(self, expected_names, specifications):
- names = [s.name for s in specifications]
- self.assertContentEqual(expected_names, names)
-
def test_anonymous_access_to_collection(self):
product = self.factory.makeProduct()
self.factory.makeSpecification(product=product, name="spec1")
self.factory.makeSpecification(product=product, name="spec2")
- # Need to endInteraction() because launchpadlib_for_anonymous() will
- # setup a new one.
- endInteraction()
- lplib = launchpadlib_for('lplib-test', person=None, version='devel')
- ws_product = ws_object(lplib, removeSecurityProxy(product))
- self.assertNamesOfSpecificationsAre(
- ["spec1", "spec2"], ws_product.all_specifications)
+ product_url = api_url(product)
+ logout()
+ webservice = LaunchpadWebServiceCaller(
+ "test", "", default_api_version="devel")
+ response = webservice.get(product_url)
+ self.assertEqual(200, response.status)
+ response = webservice.get(
+ response.jsonBody()["all_specifications_collection_link"])
+ self.assertEqual(200, response.status)
+ self.assertContentEqual(
+ ["spec1", "spec2"],
+ [entry["name"] for entry in response.jsonBody()["entries"]])
def test_product_all_specifications(self):
product = self.factory.makeProduct()
self.factory.makeSpecification(product=product, name="spec1")
self.factory.makeSpecification(product=product, name="spec2")
- product_on_webservice = self.getPillarOnWebservice(product)
- self.assertNamesOfSpecificationsAre(
- ["spec1", "spec2"], product_on_webservice.all_specifications)
+ product_url = api_url(product)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.get(product_url)
+ self.assertEqual(200, response.status)
+ response = webservice.get(
+ response.jsonBody()["all_specifications_collection_link"])
+ self.assertEqual(200, response.status)
+ self.assertContentEqual(
+ ["spec1", "spec2"],
+ [entry["name"] for entry in response.jsonBody()["entries"]])
def test_distribution_valid_specifications(self):
distribution = self.factory.makeDistribution()
@@ -380,120 +458,146 @@ class IHasSpecificationsTests(SpecificationWebserviceTestCase):
self.factory.makeSpecification(
distribution=distribution, name="spec2",
status=SpecificationDefinitionStatus.OBSOLETE)
- distro_on_webservice = self.getPillarOnWebservice(distribution)
- self.assertNamesOfSpecificationsAre(
- ["spec1"], distro_on_webservice.valid_specifications)
+ distribution_url = api_url(distribution)
+ webservice = webservice_for_person(
+ self.factory.makePerson(), default_api_version="devel")
+ response = webservice.get(distribution_url)
+ self.assertEqual(200, response.status)
+ response = webservice.get(
+ response.jsonBody()["valid_specifications_collection_link"])
+ self.assertEqual(200, response.status)
+ self.assertContentEqual(
+ ["spec1"],
+ [entry["name"] for entry in response.jsonBody()["entries"]])
-class TestSpecificationSubscription(SpecificationWebserviceTestCase):
+class TestSpecificationSubscription(TestCaseWithFactory):
- layer = AppServerLayer
+ layer = DatabaseFunctionalLayer
def test_subscribe(self):
# Test subscribe() API.
- with person_logged_in(ANONYMOUS):
- db_spec = self.factory.makeSpecification()
- db_person = self.factory.makePerson()
- launchpad = self.factory.makeLaunchpadService()
-
- spec = ws_object(launchpad, db_spec)
- person = ws_object(launchpad, db_person)
- spec.subscribe(person=person, essential=True)
- transaction.commit()
+ spec = self.factory.makeSpecification()
+ person = self.factory.makePerson()
+ spec_url = api_url(spec)
+ person_url = api_url(person)
+ webservice = webservice_for_person(
+ person, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+ response = webservice.named_post(
+ spec_url, "subscribe", person=person_url, essential=True)
+ self.assertEqual(200, response.status)
# Check the results.
- sub = db_spec.subscription(db_person)
+ login(ANONYMOUS)
+ sub = spec.subscription(person)
self.assertIsNot(None, sub)
self.assertTrue(sub.essential)
def test_unsubscribe(self):
# Test unsubscribe() API.
- with person_logged_in(ANONYMOUS):
- db_spec = self.factory.makeBlueprint()
- db_person = self.factory.makePerson()
- db_spec.subscribe(person=db_person)
- launchpad = self.factory.makeLaunchpadService(person=db_person)
-
- spec = ws_object(launchpad, db_spec)
- person = ws_object(launchpad, db_person)
- spec.unsubscribe(person=person)
- transaction.commit()
+ spec = self.factory.makeBlueprint()
+ person = self.factory.makePerson()
+ spec.subscribe(person=person)
+ spec_url = api_url(spec)
+ person_url = api_url(person)
+ webservice = webservice_for_person(
+ person, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+ response = webservice.named_post(
+ spec_url, "unsubscribe", person=person_url)
+ self.assertEqual(200, response.status)
# Check the results.
- self.assertFalse(db_spec.isSubscribed(db_person))
+ login(ANONYMOUS)
+ self.assertFalse(spec.isSubscribed(person))
def test_canBeUnsubscribedByUser(self):
# Test canBeUnsubscribedByUser() API.
- webservice = LaunchpadWebServiceCaller(
- 'launchpad-library', 'salgado-change-anything',
- domain='api.launchpad.test:8085')
-
- with person_logged_in(ANONYMOUS):
- db_spec = self.factory.makeSpecification()
- db_person = self.factory.makePerson()
- launchpad = self.factory.makeLaunchpadService()
-
- spec = ws_object(launchpad, db_spec)
- person = ws_object(launchpad, db_person)
- subscription = spec.subscribe(person=person, essential=True)
- transaction.commit()
-
- result = webservice.named_get(
- subscription['self_link'], 'canBeUnsubscribedByUser').jsonBody()
- self.assertTrue(result)
+ spec = self.factory.makeSpecification()
+ person = self.factory.makePerson()
+ with person_logged_in(person):
+ subscription = spec.subscribe(
+ person=person, subscribed_by=person, essential=True)
+ subscription_url = api_url(subscription)
+ admin_webservice = webservice_for_person(
+ getUtility(ILaunchpadCelebrities).admin.teamowner,
+ default_api_version="devel")
+ response = admin_webservice.named_get(
+ subscription_url, "canBeUnsubscribedByUser")
+ self.assertEqual(200, response.status)
+ self.assertIs(True, response.jsonBody())
-class TestSpecificationBugLinks(SpecificationWebserviceTestCase):
+class TestSpecificationBugLinks(TestCaseWithFactory):
- layer = AppServerLayer
+ layer = DatabaseFunctionalLayer
def test_bug_linking(self):
# Set up a spec, person, and bug.
- with person_logged_in(ANONYMOUS):
- db_spec = self.factory.makeSpecification()
- db_person = self.factory.makePerson()
- db_bug = self.factory.makeBug()
- launchpad = self.factory.makeLaunchpadService()
+ spec = self.factory.makeSpecification()
+ person = self.factory.makePerson()
+ bug = self.factory.makeBug()
+ spec_url = api_url(spec)
+ bug_url = api_url(bug)
+ webservice = webservice_for_person(
+ person, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+
+ # There are no bugs associated with the spec/blueprint yet.
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ spec_bugs_url = response.jsonBody()["bugs_collection_link"]
+ response = webservice.get(spec_bugs_url)
+ self.assertEqual(200, response.status)
+ self.assertEqual(0, response.jsonBody()["total_size"])
# Link the bug to the spec via the web service.
- with person_logged_in(db_person):
- spec = ws_object(launchpad, db_spec)
- bug = ws_object(launchpad, db_bug)
- # There are no bugs associated with the spec/blueprint yet.
- self.assertEqual(0, spec.bugs.total_size)
- spec.linkBug(bug=bug)
- transaction.commit()
+ response = webservice.named_post(spec_url, "linkBug", bug=bug_url)
+ self.assertEqual(200, response.status)
# The spec now has one bug associated with it and that bug is the one
# we linked.
- self.assertEqual(1, spec.bugs.total_size)
- self.assertEqual(bug.id, spec.bugs[0].id)
+ response = webservice.get(spec_bugs_url)
+ self.assertEqual(200, response.status)
+ self.assertThat(response.jsonBody(), ContainsDict({
+ "total_size": Equals(1),
+ "entries": MatchesListwise([ContainsDict({"id": Equals(bug.id)})]),
+ }))
def test_bug_unlinking(self):
# Set up a spec, person, and bug, then link the bug to the spec.
- with person_logged_in(ANONYMOUS):
- db_spec = self.factory.makeBlueprint()
- db_person = self.factory.makePerson()
- db_bug = self.factory.makeBug()
- launchpad = self.factory.makeLaunchpadService(person=db_person)
-
- spec = ws_object(launchpad, db_spec)
- bug = ws_object(launchpad, db_bug)
- spec.linkBug(bug=bug)
+ spec = self.factory.makeBlueprint()
+ person = self.factory.makePerson()
+ bug = self.factory.makeBug()
+ spec_url = api_url(spec)
+ bug_url = api_url(bug)
+ with person_logged_in(spec.owner):
+ spec.linkBug(bug)
+ webservice = webservice_for_person(
+ person, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
# There is only one bug linked at the moment.
- self.assertEqual(1, spec.bugs.total_size)
+ response = webservice.get(spec_url)
+ self.assertEqual(200, response.status)
+ spec_bugs_url = response.jsonBody()["bugs_collection_link"]
+ response = webservice.get(spec_bugs_url)
+ self.assertEqual(200, response.status)
+ self.assertEqual(1, response.jsonBody()["total_size"])
- spec.unlinkBug(bug=bug)
- transaction.commit()
+ response = webservice.named_post(spec_url, "unlinkBug", bug=bug_url)
+ self.assertEqual(200, response.status)
# Now that we've unlinked the bug, there are no linked bugs at all.
- self.assertEqual(0, spec.bugs.total_size)
+ response = webservice.get(spec_bugs_url)
+ self.assertEqual(200, response.status)
+ self.assertEqual(0, response.jsonBody()["total_size"])
-class TestSpecificationGoalHandling(SpecificationWebserviceTestCase):
+class TestSpecificationGoalHandling(TestCaseWithFactory):
- layer = AppServerLayer
+ layer = DatabaseFunctionalLayer
def setUp(self):
super(TestSpecificationGoalHandling, self).setUp()
@@ -501,53 +605,64 @@ class TestSpecificationGoalHandling(SpecificationWebserviceTestCase):
self.proposer = self.factory.makePerson()
self.product = self.factory.makeProduct(driver=self.driver)
self.series = self.factory.makeProductSeries(product=self.product)
+ self.series_url = api_url(self.series)
def test_goal_propose_and_accept(self):
# Webservice clients can propose and accept spec series goals.
- db_spec = self.factory.makeBlueprint(product=self.product,
- owner=self.proposer)
+ spec = self.factory.makeBlueprint(
+ product=self.product, owner=self.proposer)
+ spec_url = api_url(spec)
+
# Propose for series goal
+ proposer_webservice = webservice_for_person(
+ self.proposer, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+ response = proposer_webservice.named_post(
+ spec_url, "proposeGoal", goal=self.series_url)
+ self.assertEqual(200, response.status)
with person_logged_in(self.proposer):
- launchpad = self.factory.makeLaunchpadService(person=self.proposer)
- spec = ws_object(launchpad, db_spec)
- series = ws_object(launchpad, self.series)
- spec.proposeGoal(goal=series)
- transaction.commit()
- self.assertEqual(db_spec.goal, self.series)
+ self.assertEqual(spec.goal, self.series)
self.assertFalse(spec.has_accepted_goal)
# Accept series goal
+ driver_webservice = webservice_for_person(
+ self.driver, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+ response = driver_webservice.named_post(spec_url, "acceptGoal")
+ self.assertEqual(200, response.status)
with person_logged_in(self.driver):
- launchpad = self.factory.makeLaunchpadService(person=self.driver)
- spec = ws_object(launchpad, db_spec)
- spec.acceptGoal()
- transaction.commit()
self.assertTrue(spec.has_accepted_goal)
def test_goal_propose_decline_and_clear(self):
# Webservice clients can decline and clear spec series goals.
- db_spec = self.factory.makeBlueprint(product=self.product,
- owner=self.proposer)
+ spec = self.factory.makeBlueprint(
+ product=self.product, owner=self.proposer)
+ spec_url = api_url(spec)
+
# Propose for series goal
+ proposer_webservice = webservice_for_person(
+ self.proposer, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+ response = proposer_webservice.named_post(
+ spec_url, "proposeGoal", goal=self.series_url)
+ self.assertEqual(200, response.status)
with person_logged_in(self.proposer):
- launchpad = self.factory.makeLaunchpadService(person=self.proposer)
- spec = ws_object(launchpad, db_spec)
- series = ws_object(launchpad, self.series)
- spec.proposeGoal(goal=series)
- transaction.commit()
- self.assertEqual(db_spec.goal, self.series)
+ self.assertEqual(spec.goal, self.series)
self.assertFalse(spec.has_accepted_goal)
+ # Decline series goal
+ driver_webservice = webservice_for_person(
+ self.driver, permission=OAuthPermission.WRITE_PUBLIC,
+ default_api_version="devel")
+ response = driver_webservice.named_post(spec_url, "declineGoal")
+ self.assertEqual(200, response.status)
with person_logged_in(self.driver):
- # Decline series goal
- launchpad = self.factory.makeLaunchpadService(person=self.driver)
- spec = ws_object(launchpad, db_spec)
- spec.declineGoal()
- transaction.commit()
self.assertFalse(spec.has_accepted_goal)
- self.assertEqual(db_spec.goal, self.series)
+ self.assertEqual(spec.goal, self.series)
- # Clear series goal as a driver
- spec.proposeGoal(goal=None)
- transaction.commit()
- self.assertIsNone(db_spec.goal)
+ # Clear series goal as a driver
+ response = driver_webservice.named_post(
+ spec_url, "proposeGoal", goal=None)
+ self.assertEqual(200, response.status)
+ with person_logged_in(self.driver):
+ self.assertIsNone(spec.goal)
diff --git a/lib/lp/testing/pages.py b/lib/lp/testing/pages.py
index cff8368..b24f141 100644
--- a/lib/lp/testing/pages.py
+++ b/lib/lp/testing/pages.py
@@ -158,7 +158,8 @@ class LaunchpadWebServiceCaller(WebServiceCaller):
def __init__(self, oauth_consumer_key=None, oauth_access_key=None,
oauth_access_secret=None, handle_errors=True,
- domain='api.launchpad.test', protocol='http'):
+ domain='api.launchpad.test', protocol='http',
+ default_api_version=None):
"""Create a LaunchpadWebServiceCaller.
:param oauth_consumer_key: The OAuth consumer key to use.
:param oauth_access_key: The OAuth access key to use for the request.
@@ -184,6 +185,8 @@ class LaunchpadWebServiceCaller(WebServiceCaller):
self.consumer = None
self.access_token = None
self.handle_errors = handle_errors
+ if default_api_version is not None:
+ self.default_api_version = default_api_version
WebServiceCaller.__init__(self, handle_errors, domain, protocol)
default_api_version = "beta"
@@ -745,7 +748,7 @@ def safe_canonical_url(*args, **kwargs):
def webservice_for_person(person, consumer_key=u'launchpad-library',
permission=OAuthPermission.READ_PUBLIC,
- context=None):
+ context=None, default_api_version=None):
"""Return a valid LaunchpadWebServiceCaller for the person.
Use this method to create a way to test the webservice that doesn't depend
@@ -763,7 +766,8 @@ def webservice_for_person(person, consumer_key=u'launchpad-library',
access_token, access_secret = request_token.createAccessToken()
logout()
service = LaunchpadWebServiceCaller(
- consumer_key, access_token.key, access_secret)
+ consumer_key, access_token.key, access_secret,
+ default_api_version=default_api_version)
service.user = person
return service
