¡@
OpenStack Study: workflows.py**** CubicPower OpenStack Study ****
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # Copyright 2012 Nebula, Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. from django import forms from django import http from horizon import exceptions from horizon.test import helpers as test from horizon import workflows PROJECT_ID = "a23lkjre389fwenj" INSTANCE_ID = "sdlkjhf9832roiw"
**** CubicPower OpenStack Study **** def local_callback_func(request, context): return "one"
**** CubicPower OpenStack Study **** def other_callback_func(request, context): return "two"
**** CubicPower OpenStack Study **** def extra_callback_func(request, context): return "extra"
**** CubicPower OpenStack Study **** class TestActionOne(workflows.Action): project_id = forms.ChoiceField(label="Project") user_id = forms.ChoiceField(label="User") class Meta: name = "Test Action One" slug = "test_action_one"
**** CubicPower OpenStack Study **** def populate_project_id_choices(self, request, context): return [(PROJECT_ID, "test_project")]
**** CubicPower OpenStack Study **** def populate_user_id_choices(self, request, context): return [(request.user.id, request.user.username)]
**** CubicPower OpenStack Study **** def handle(self, request, context): return {"foo": "bar"}
**** CubicPower OpenStack Study **** class TestActionTwo(workflows.Action): instance_id = forms.CharField(label="Instance") class Meta: name = "Test Action Two" slug = "test_action_two"
**** CubicPower OpenStack Study **** class TestActionThree(workflows.Action): extra = forms.CharField(widget=forms.widgets.Textarea) class Meta: name = "Test Action Three" slug = "test_action_three"
**** CubicPower OpenStack Study **** class AdminAction(workflows.Action): admin_id = forms.CharField(label="Admin") class Meta: name = "Admin Action" slug = "admin_action" permissions = ("horizon.test",)
**** CubicPower OpenStack Study **** class TestStepOne(workflows.Step): action_class = TestActionOne contributes = ("project_id", "user_id")
**** CubicPower OpenStack Study **** class TestStepTwo(workflows.Step): action_class = TestActionTwo depends_on = ("project_id",) contributes = ("instance_id",) connections = {"project_id": (local_callback_func, "horizon.test.tests.workflows.other_callback_func")}
**** CubicPower OpenStack Study **** class TestExtraStep(workflows.Step): action_class = TestActionThree depends_on = ("project_id",) contributes = ("extra_data",) connections = {"project_id": (extra_callback_func,)} after = TestStepOne before = TestStepTwo
**** CubicPower OpenStack Study **** class AdminStep(workflows.Step): action_class = AdminAction contributes = ("admin_id",) after = TestStepOne before = TestStepTwo
**** CubicPower OpenStack Study **** class TestWorkflow(workflows.Workflow): slug = "test_workflow"
**** CubicPower OpenStack Study **** class TestWorkflowView(workflows.WorkflowView): workflow_class = TestWorkflow template_name = "workflow.html"
**** CubicPower OpenStack Study **** class TestFullscreenWorkflow(workflows.Workflow): slug = 'test_fullscreen_workflow'
**** CubicPower OpenStack Study **** class TestFullscreenWorkflowView(workflows.WorkflowView): workflow_class = TestFullscreenWorkflow template_name = "workflow.html"
**** CubicPower OpenStack Study **** class WorkflowsTests(test.TestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(WorkflowsTests, self).setUp()
**** CubicPower OpenStack Study **** def tearDown(self): super(WorkflowsTests, self).tearDown() self._reset_workflow()
**** CubicPower OpenStack Study **** def _reset_workflow(self): TestWorkflow._cls_registry = set([])
**** CubicPower OpenStack Study **** def test_workflow_construction(self): TestWorkflow.register(TestExtraStep) flow = TestWorkflow(self.request) self.assertQuerysetEqual(flow.steps, [' ' ' self.assertEqual(flow.depends_on, set(['project_id']))
**** CubicPower OpenStack Study **** def test_step_construction(self): step_one = TestStepOne(TestWorkflow(self.request)) # Action slug is moved from Meta by metaclass, and # Step inherits slug from action. self.assertEqual(step_one.name, TestActionOne.name) self.assertEqual(step_one.slug, TestActionOne.slug) # Handlers should be empty since there are no connections. self.assertEqual(step_one._handlers, {}) step_two = TestStepTwo(TestWorkflow(self.request)) # Handlers should be populated since we do have connections. self.assertEqual(step_two._handlers["project_id"], [local_callback_func, other_callback_func])
**** CubicPower OpenStack Study **** def test_step_invalid_callback(self): # This should raise an exception class InvalidStep(TestStepTwo): connections = {"project_id": ('local_callback_func',)} with self.assertRaises(ValueError): InvalidStep(TestWorkflow(self.request))
**** CubicPower OpenStack Study **** def test_connection_handlers_called(self): TestWorkflow.register(TestExtraStep) flow = TestWorkflow(self.request) # This should set the value without any errors, but trigger nothing flow.context['does_not_exist'] = False self.assertEqual(flow.context['does_not_exist'], False) # The order here is relevant. Note that we inserted "extra" between # steps one and two, and one has no handlers, so we should see # a response from extra, then one from each of step two's handlers. val = flow.context.set('project_id', PROJECT_ID) self.assertEqual(val, [('test_action_three', 'extra'), ('test_action_two', 'one'), ('test_action_two', 'two')])
**** CubicPower OpenStack Study **** def test_workflow_validation(self): flow = TestWorkflow(self.request) # Missing items fail validation. with self.assertRaises(exceptions.WorkflowValidationError): flow.is_valid() # All required items pass validation. seed = {"project_id": PROJECT_ID, "user_id": self.user.id, "instance_id": INSTANCE_ID} req = self.factory.post("/", seed) req.user = self.user flow = TestWorkflow(req, context_seed={"project_id": PROJECT_ID}) for step in flow.steps: if not step.action.is_valid(): self.fail("Step %s was unexpectedly invalid: %s" % (step.slug, step.action.errors)) self.assertTrue(flow.is_valid()) # Additional items shouldn't affect validation flow.context.set("extra_data", "foo") self.assertTrue(flow.is_valid())
**** CubicPower OpenStack Study **** def test_workflow_finalization(self): flow = TestWorkflow(self.request) self.assertTrue(flow.finalize())
**** CubicPower OpenStack Study **** def test_workflow_view(self): view = TestWorkflowView.as_view() req = self.factory.get("/") res = view(req) self.assertEqual(res.status_code, 200)
**** CubicPower OpenStack Study **** def test_workflow_registration(self): req = self.factory.get("/foo") flow = TestWorkflow(req) self.assertQuerysetEqual(flow.steps, [' ' TestWorkflow.register(TestExtraStep) flow = TestWorkflow(req) self.assertQuerysetEqual(flow.steps, [' ' '
**** CubicPower OpenStack Study **** def test_workflow_render(self): TestWorkflow.register(TestExtraStep) req = self.factory.get("/foo") flow = TestWorkflow(req) output = http.HttpResponse(flow.render()) self.assertContains(output, unicode(flow.name)) self.assertContains(output, unicode(TestActionOne.name)) self.assertContains(output, unicode(TestActionTwo.name)) self.assertContains(output, unicode(TestActionThree.name))
**** CubicPower OpenStack Study **** def test_has_permissions(self): self.assertQuerysetEqual(TestWorkflow._cls_registry, []) TestWorkflow.register(AdminStep) flow = TestWorkflow(self.request) step = AdminStep(flow) self.assertItemsEqual(step.permissions, ("horizon.test",)) self.assertQuerysetEqual(flow.steps, [' ' self.set_permissions(['test']) self.request.user = self.user flow = TestWorkflow(self.request) self.assertQuerysetEqual(flow.steps, [' ' '
**** CubicPower OpenStack Study **** def test_entry_point(self): req = self.factory.get("/foo") flow = TestWorkflow(req) self.assertEqual(flow.get_entry_point(), "test_action_one") flow = TestWorkflow(req, entry_point="test_action_two") self.assertEqual(flow.get_entry_point(), "test_action_two")
**** CubicPower OpenStack Study **** def test_fullscreenworkflow_view(self): view = TestFullscreenWorkflowView.as_view() req = self.factory.get("/") req.is_ajax = lambda: True res = view(req) output = res.render() self.assertRegexpMatches(str(output), 'class="[^"]*\\bfullscreen\\b[^"]*"')
**** CubicPower OpenStack Study **** def test_notfullscreenworkflow_view(self): view = TestWorkflowView.as_view() req = self.factory.get("/") req.is_ajax = lambda: True res = view(req) output = res.render() self.assertNotRegexpMatches(str(output), 'class="[^"]*\\bfullscreen\\b[^"]*"')
|