| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831 |
- #!/usr/bin/env python
- # Licensed to Cloudera, Inc. under one
- # or more contributor license agreements. See the NOTICE file
- # distributed with this work for additional information
- # regarding copyright ownership. Cloudera, Inc. licenses this file
- # to you 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.
- try:
- import json
- except ImportError:
- import simplejson as json
- import logging
- import re
- import os
- from nose.plugins.skip import SkipTest
- from nose.tools import assert_true, assert_false, assert_equal, assert_not_equal
- from django.contrib.auth.models import User
- from django.core.urlresolvers import reverse
- from desktop.lib.django_test_util import make_logged_in_client
- from desktop.lib.test_utils import grant_access, add_permission
- from jobsub.management.commands import jobsub_setup
- from jobsub.models import OozieDesign
- from liboozie import oozie_api
- from liboozie.conf import OOZIE_URL
- from liboozie.oozie_api_test import OozieServerProvider
- from liboozie.types import WorkflowList, Workflow as OozieWorkflow, Coordinator as OozieCoordinator,\
- CoordinatorList, WorkflowAction
- from oozie.models import Workflow, Node, Kill, Link, Job, Coordinator, History,\
- find_parameters, NODE_TYPES
- from oozie.conf import SHARE_JOBS
- from oozie.utils import workflow_to_dict, model_to_dict
- from oozie.import_workflow import import_workflow
- LOG = logging.getLogger(__name__)
- _INITIALIZED = False
- class MockOozieApi:
- JSON_WORKFLOW_LIST = [{u'status': u'RUNNING', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:35:48 GMT', u'appName': u'WordCount1', u'lastModTime': u'Mon, 30 Jul 2012 22:37:00 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': 'job_201208072118_0044', u'consoleUrl': u'http://runreal:11000/oozie?job=0000012-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:35:48 GMT', u'toString': u'Workflow id[0000012-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:37:00 GMT', u'id': u'0000012-120725142744176-oozie-oozi-W', u'group': None, u'user': u'test'},
- {u'status': u'KILLED', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:31:08 GMT', u'appName': u'WordCount2', u'lastModTime': u'Mon, 30 Jul 2012 22:32:20 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': '-', u'consoleUrl': u'http://runreal:11000/oozie?job=0000011-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:31:08 GMT', u'toString': u'Workflow id[0000011-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:32:20 GMT', u'id': u'0000011-120725142744176-oozie-oozi-W', u'group': None, u'user': u'test'},
- {u'status': u'SUCCEEDED', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:20:48 GMT', u'appName': u'WordCount3', u'lastModTime': u'Mon, 30 Jul 2012 22:22:00 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': '', u'consoleUrl': u'http://runreal:11000/oozie?job=0000009-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:20:48 GMT', u'toString': u'Workflow id[0000009-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:22:00 GMT', u'id': u'0000009-120725142744176-oozie-oozi-W', u'group': None, u'user': u'test'},
- {u'status': u'SUCCEEDED', u'run': 0, u'startTime': u'Mon, 30 Jul 2012 22:16:58 GMT', u'appName': u'WordCount4', u'lastModTime': u'Mon, 30 Jul 2012 22:18:10 GMT', u'actions': [], u'acl': None, u'appPath': None, u'externalId': None, u'consoleUrl': u'http://runreal:11000/oozie?job=0000008-120725142744176-oozie-oozi-W', u'conf': None, u'parentId': None, u'createdTime': u'Mon, 30 Jul 2012 22:16:58 GMT', u'toString': u'Workflow id[0000008-120725142744176-oozie-oozi-W] status[SUCCEEDED]', u'endTime': u'Mon, 30 Jul 2012 22:18:10 GMT', u'id': u'0000008-120725142744176-oozie-oozi-W', u'group': None, u'user': u'test'}]
- WORKFLOW_IDS = [wf['id'] for wf in JSON_WORKFLOW_LIST]
- WORKFLOW_DICT = dict([(wf['id'], wf) for wf in JSON_WORKFLOW_LIST])
- JSON_COORDINATOR_LIST = [{u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000041-120717205528122-oozie-oozi-C] status[RUNNING]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'RUNNING', u'conf': None, u'user': u'test', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/test/demo2', u'timeUnit': u'DAY', u'coordJobId': u'0000041-120717205528122-oozie-oozi-C', u'coordJobName': u'DailyWordCount1', u'nextMaterializedTime': u'Wed, 04 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Wed, 04 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 00:00:00 GMT'},
- {u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000011-120706144403213-oozie-oozi-C] status[DONEWITHERROR]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'DONEWITHERROR', u'conf': None, u'user': u'test', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/hue/jobsub/_romain_-design-2', u'timeUnit': u'DAY', u'coordJobId': u'0000011-120706144403213-oozie-oozi-C', u'coordJobName': u'DailyWordCount2', u'nextMaterializedTime': u'Thu, 05 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Thu, 05 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 18:54:00 GMT'},
- {u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000010-120706144403213-oozie-oozi-C] status[DONEWITHERROR]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'DONEWITHERROR', u'conf': None, u'user': u'test', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/hue/jobsub/_romain_-design-2', u'timeUnit': u'DAY', u'coordJobId': u'0000010-120706144403213-oozie-oozi-C', u'coordJobName': u'DailyWordCount3', u'nextMaterializedTime': u'Thu, 05 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Thu, 05 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 18:54:00 GMT'},
- {u'startTime': u'Sun, 01 Jul 2012 00:00:00 GMT', u'actions': [], u'frequency': 1, u'concurrency': 1, u'pauseTime': None, u'group': None, u'toString': u'Coornidator application id[0000009-120706144403213-oozie-oozi-C] status[DONEWITHERROR]', u'consoleUrl': None, u'mat_throttling': 0, u'status': u'DONEWITHERROR', u'conf': None, u'user': u'test', u'timeOut': 120, u'coordJobPath': u'hdfs://localhost:8020/user/hue/jobsub/_romain_-design-2', u'timeUnit': u'DAY', u'coordJobId': u'0000009-120706144403213-oozie-oozi-C', u'coordJobName': u'DailyWordCount4', u'nextMaterializedTime': u'Thu, 05 Jul 2012 00:00:00 GMT', u'coordExternalId': None, u'acl': None, u'lastAction': u'Thu, 05 Jul 2012 00:00:00 GMT', u'executionPolicy': u'FIFO', u'timeZone': u'America/Los_Angeles', u'endTime': u'Wed, 04 Jul 2012 18:54:00 GMT'}]
- COORDINATOR_IDS = [coord['coordJobId'] for coord in JSON_COORDINATOR_LIST]
- COORDINATOR_DICT = dict([(coord['coordJobId'], coord) for coord in JSON_COORDINATOR_LIST])
- WORKFLOW_ACTION = {u'status': u'OK', u'retries': 0, u'transition': u'end', u'stats': None, u'startTime': u'Fri, 10 Aug 2012 05:24:21 GMT', u'toString': u'Action name[WordCount] status[OK]', u'cred': u'null', u'errorMessage': None, u'errorCode': None, u'consoleUrl': u'http://localhost:50030/jobdetails.jsp?jobid=job_201208072118_0044', u'externalId': u'job_201208072118_0044', u'externalStatus': u'SUCCEEDED', u'conf': u'<map-reduce xmlns="uri:oozie:workflow:0.2">\r\n <job-tracker>localhost:8021</job-tracker>\r\n <name-node>hdfs://localhost:8020</name-node>\r\n <configuration>\r\n <property>\r\n <name>mapred.mapper.regex</name>\r\n <value>dream</value>\r\n </property>\r\n <property>\r\n <name>mapred.input.dir</name>\r\n <value>/user/test/words/20120702</value>\r\n </property>\r\n <property>\r\n <name>mapred.output.dir</name>\r\n <value>/user/test/out/rrwords/20120702</value>\r\n </property>\r\n <property>\r\n <name>mapred.mapper.class</name>\r\n <value>org.apache.hadoop.mapred.lib.RegexMapper</value>\r\n </property>\r\n <property>\r\n <name>mapred.combiner.class</name>\r\n <value>org.apache.hadoop.mapred.lib.LongSumReducer</value>\r\n </property>\r\n <property>\r\n <name>mapred.reducer.class</name>\r\n <value>org.apache.hadoop.mapred.lib.LongSumReducer</value>\r\n </property>\r\n <property>\r\n <name>mapred.output.key.class</name>\r\n <value>org.apache.hadoop.io.Text</value>\r\n </property>\r\n <property>\r\n <name>mapred.output.value.class</name>\r\n <value>org.apache.hadoop.io.LongWritable</value>\r\n </property>\r\n </configuration>\r\n</map-reduce>', u'type': u'map-reduce', u'trackerUri': u'localhost:8021', u'externalChildIDs': None, u'endTime': u'Fri, 10 Aug 2012 05:24:38 GMT', u'data': None, u'id': u'0000012-120725142744176-oozie-oozi-W@WordCount', u'name': u'WordCount'}
- def __init__(self, *args, **kwargs):
- pass
- def setuser(self, user):
- pass
- def submit_job(self, properties):
- return 'ONE-OOZIE-ID-W'
- def get_workflows(self, **kwargs):
- workflows = MockOozieApi.JSON_WORKFLOW_LIST
- if 'user' in kwargs:
- workflows = filter(lambda wf: wf['user'] == kwargs['user'], workflows)
- return WorkflowList(self, {'offset': 0, 'total': 4, 'workflows': workflows})
- def get_coordinators(self, **kwargs):
- coordinatorjobs = MockOozieApi.JSON_COORDINATOR_LIST
- if 'user' in kwargs:
- coordinatorjobs = filter(lambda coord: coord['user'] == kwargs['user'], coordinatorjobs)
- return CoordinatorList(self, {'offset': 0, 'total': 5, 'coordinatorjobs': coordinatorjobs})
- def get_job(self, job_id):
- if job_id in MockOozieApi.WORKFLOW_DICT:
- return OozieWorkflow(self, MockOozieApi.WORKFLOW_DICT[job_id])
- else:
- return OozieWorkflow(self, {'id': job_id, 'actions': []})
- def get_coordinator(self, job_id):
- if job_id in MockOozieApi.COORDINATOR_DICT:
- return OozieCoordinator(self, MockOozieApi.COORDINATOR_DICT[job_id])
- else:
- return OozieCoordinator(self, {'id': job_id, 'actions': []})
- def get_action(self, action_id):
- return WorkflowAction(MockOozieApi.WORKFLOW_ACTION)
- def job_control(self, job_id, action):
- return 'Done'
- def get_job_definition(self, jobid):
- return '<xml></xml>'
- def get_job_log(self, jobid):
- return '<xml></xml>'
- class OozieMockBase(object):
- def setUp(self):
- # Beware: Monkey patch Oozie/LibOozie with Mock API
- if not hasattr(oozie_api, 'OriginalOozieApi'):
- oozie_api.OriginalOozieApi = oozie_api.OozieApi
- if not hasattr(Workflow.objects, 'original_check_workspace'):
- Workflow.objects.original_check_workspace = Workflow.objects.check_workspace
- Workflow.objects.check_workspace = lambda a, b: None
- oozie_api.OozieApi = MockOozieApi
- oozie_api._api_cache = None
- Coordinator.objects.all().delete()
- self.c = make_logged_in_client(is_superuser=False)
- grant_access("test", "test", "oozie")
- self.user = User.objects.get(username='test')
- self.wf = create_workflow(self.c)
- def tearDown(self):
- oozie_api.OozieApi = oozie_api.OriginalOozieApi
- Workflow.objects.check_workspace = Workflow.objects.original_check_workspace
- oozie_api._api_cache = None
- def setup_simple_workflow(self):
- """ Creates a linear workflow """
- Link.objects.filter(parent__workflow=self.wf).delete()
- Link(parent=self.wf.start, child=self.wf.end, name="related").save()
- action1 = add_node(self.wf, 'action-name-1', 'mapreduce', [self.wf.start], {
- 'description': '',
- 'files': '[]',
- 'jar_path': '/user/hue/oozie/examples/lib/hadoop-examples.jar',
- 'job_properties': '[{"name":"sleep","value":"${SLEEP}"}]',
- 'prepares': '[{"value":"${output}","type":"delete"},{"value":"/test","type":"mkdir"}]',
- 'archives': '[]',
- })
- action2 = add_node(self.wf, 'action-name-2', 'mapreduce', [action1], {
- 'description': '',
- 'files': '[]',
- 'jar_path': '/user/hue/oozie/examples/lib/hadoop-examples.jar',
- 'job_properties': '[{"name":"sleep","value":"${SLEEP}"}]',
- 'prepares': '[{"value":"${output}","type":"delete"},{"value":"/test","type":"mkdir"}]',
- 'archives': '[]',
- })
- action3 = add_node(self.wf, 'action-name-3', 'mapreduce', [action2], {
- 'description': '',
- 'files': '[]',
- 'jar_path': '/user/hue/oozie/examples/lib/hadoop-examples.jar',
- 'job_properties': '[{"name":"sleep","value":"${SLEEP}"}]',
- 'prepares': '[{"value":"${output}","type":"delete"},{"value":"/test","type":"mkdir"}]',
- 'archives': '[]',
- })
- Link(parent=action3, child=self.wf.end, name="ok").save()
- def setup_forking_workflow(self):
- """ Creates a workflow with a fork """
- Link.objects.filter(parent__workflow=self.wf).delete()
- Link(parent=self.wf.start, child=self.wf.end, name="related").save()
- fork1 = add_node(self.wf, 'fork-name-1', 'fork', [self.wf.start])
- action1 = add_node(self.wf, 'action-name-1', 'mapreduce', [fork1])
- action2 = add_node(self.wf, 'action-name-2', 'mapreduce', [fork1])
- join1 = add_node(self.wf, 'join-name-1', 'join', [action1, action2])
- Link(parent=fork1, child=join1, name="related").save()
- action3 = add_node(self.wf, 'action-name-3', 'mapreduce', [join1])
- Link(parent=action3, child=self.wf.end, name="ok").save()
- class OozieBase(OozieServerProvider):
- requires_hadoop = True
- def setUp(self):
- OozieServerProvider.setup_class()
- self.c = make_logged_in_client(is_superuser=False)
- self.user = User.objects.get(username="test")
- grant_access("test", "test", "oozie")
- self.cluster = OozieServerProvider.cluster
- self.install_examples()
- # Ensure access to MR folder
- self.cluster.fs.do_as_superuser(self.cluster.fs.chmod, '/tmp', 0777, recursive=True)
- def install_examples(self):
- global _INITIALIZED
- if _INITIALIZED:
- return
- self.c.post(reverse('oozie:setup_app'))
- self.cluster.fs.do_as_user('test', self.cluster.fs.create_home_dir, '/user/test')
- self.cluster.fs.do_as_superuser(self.cluster.fs.chmod, '/user/test', 0777, True)
- hue = User.objects.create_user('hue', 'hue' + '@localhost', 'hue')
- Workflow.objects.update(owner=hue)
- _INITIALIZED = True
- def setup_simple_workflow(self):
- """ Creates a linear workflow """
- Link.objects.filter(parent__workflow=self.wf).delete()
- Link(parent=self.wf.start, child=self.wf.end, name="related").save()
- action1 = add_node(self.wf, 'action-name-1', 'mapreduce', [self.wf.start], {
- 'description': '',
- 'files': '[]',
- 'jar_path': '/user/hue/oozie/examples/lib/hadoop-examples.jar',
- 'job_properties': '[{"name":"sleep","value":"${SLEEP}"}]',
- 'prepares': '[{"value":"${output}","type":"delete"},{"value":"/test","type":"mkdir"}]',
- 'archives': '[]',
- })
- action2 = add_node(self.wf, 'action-name-2', 'mapreduce', [action1], {
- 'description': '',
- 'files': '[]',
- 'jar_path': '/user/hue/oozie/examples/lib/hadoop-examples.jar',
- 'job_properties': '[{"name":"sleep","value":"${SLEEP}"}]',
- 'prepares': '[{"value":"${output}","type":"delete"},{"value":"/test","type":"mkdir"}]',
- 'archives': '[]',
- })
- action3 = add_node(self.wf, 'action-name-3', 'mapreduce', [action2], {
- 'description': '',
- 'files': '[]',
- 'jar_path': '/user/hue/oozie/examples/lib/hadoop-examples.jar',
- 'job_properties': '[{"name":"sleep","value":"${SLEEP}"}]',
- 'prepares': '[{"value":"${output}","type":"delete"},{"value":"/test","type":"mkdir"}]',
- 'archives': '[]',
- })
- Link(parent=action3, child=self.wf.end, name="ok").save()
- def setup_forking_workflow(self):
- """ Creates a workflow with a fork """
- Link.objects.filter(parent__workflow=self.wf).delete()
- Link(parent=self.wf.start, child=self.wf.end, name="related").save()
- fork1 = add_node(self.wf, 'fork-name-1', 'fork', [self.wf.start])
- action1 = add_node(self.wf, 'action-name-1', 'mapreduce', [fork1])
- action2 = add_node(self.wf, 'action-name-2', 'mapreduce', [fork1])
- join1 = add_node(self.wf, 'join-name-1', 'join', [action1, action2])
- Link(parent=fork1, child=join1, name="related").save()
- action3 = add_node(self.wf, 'action-name-3', 'mapreduce', [join1])
- Link(parent=action3, child=self.wf.end, name="ok").save()
- class TestAPI(OozieMockBase):
- def setUp(self):
- OozieMockBase.setUp(self)
- # When updating wf, update wf_json as well!
- self.wf = Workflow.objects.get(name='wf-name-1')
- def test_workflow_save(self):
- self.setup_simple_workflow()
- workflow_dict = workflow_to_dict(self.wf)
- workflow_dict = remove_related_fields( workflow_dict )
- workflow_json = json.dumps(workflow_dict)
- response = self.c.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json})
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(0, test_response_json_object['status'])
- # Change property and save
- workflow_dict = workflow_to_dict(self.wf)
- workflow_dict = remove_related_fields( workflow_dict )
- workflow_dict['description'] = 'test'
- workflow_json = json.dumps(workflow_dict)
- response = self.c.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json})
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(0, test_response_json_object['status'])
- wf = Workflow.objects.get(id=self.wf.id)
- assert_equal('test', wf.description)
- assert_equal(self.wf.name, wf.name)
- # Change node and save
- workflow_dict = workflow_to_dict(self.wf)
- workflow_dict = remove_related_fields( workflow_dict )
- workflow_dict['nodes'][2]['name'] = 'new-name'
- node_id = workflow_dict['nodes'][2]['id']
- workflow_json = json.dumps(workflow_dict)
- response = self.c.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json})
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(0, test_response_json_object['status'])
- node = Node.objects.get(id=node_id)
- assert_equal('new-name', node.name)
- def test_workflow_save_fail(self):
- self.setup_simple_workflow()
- # Bad workflow name
- workflow_dict = workflow_to_dict(self.wf)
- del workflow_dict['name']
- workflow_dict = remove_related_fields( workflow_dict )
- workflow_json = json.dumps(workflow_dict)
- response = self.c.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- assert_equal(400, response.status_code)
- # Bad node name
- workflow_dict = workflow_to_dict(self.wf)
- del workflow_dict['nodes'][2]['name']
- workflow_dict = remove_related_fields( workflow_dict )
- workflow_json = json.dumps(workflow_dict)
- response = self.c.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- assert_equal(400, response.status_code)
- def test_workflow(self):
- response = self.c.get(reverse('oozie:workflow', kwargs={'workflow': self.wf.pk}))
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(0, test_response_json_object['status'])
- def test_workflow_validate_action(self):
- data = {"files":"[\"hive-site.xml\"]","job_xml":"hive-site.xml","description":"Show databases","workflow":17,"child_links":[{"comment":"","name":"ok","id":106,"parent":76,"child":74},{"comment":"","name":"error","id":107,"parent":76,"child":73}],"job_properties":"[{\"name\":\"oozie.hive.defaults\",\"value\":\"hive-site.xml\"}]","node_type":"hive","params":"[{\"value\":\"INPUT=/user/hue/oozie/workspaces/data\",\"type\":\"param\"}]","archives":"[]","node_ptr":76,"prepares":"[]","script_path":"hive.sql","id":76,"name":"Hive"}
- response = self.c.post(reverse('oozie:workflow_validate_action', kwargs={'workflow': self.wf.pk, 'node_type': 'hive'}), data={'node': json.dumps(data)}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(0, test_response_json_object['status'])
- assert_true('name' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['name']), test_response_json_object['data'])
- assert_true('description' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['description']), test_response_json_object['data'])
- assert_true('script_path' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['script_path']), test_response_json_object['data'])
- assert_true('job_xml' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['job_xml']), test_response_json_object['data'])
- assert_true('job_properties' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['job_properties']), test_response_json_object['data'])
- assert_true('files' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['files']), test_response_json_object['data'])
- assert_true('params' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['params']), test_response_json_object['data'])
- assert_true('prepares' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['prepares']), test_response_json_object['data'])
- assert_true('archives' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['archives']), test_response_json_object['data'])
- def test_workflow_validate_action_fail(self):
- # Empty files field
- data = {"job_xml":"hive-site.xml","description":"Show databases","workflow":17,"child_links":[{"comment":"","name":"ok","id":106,"parent":76,"child":74},{"comment":"","name":"error","id":107,"parent":76,"child":73}],"job_properties":"[{\"name\":\"oozie.hive.defaults\",\"value\":\"hive-site.xml\"}]","node_type":"hive","params":"[{\"value\":\"INPUT=/user/hue/oozie/workspaces/data\",\"type\":\"param\"}]","archives":"[]","node_ptr":76,"prepares":"[]","script_path":"hive.sql","id":76,"name":"Hive"}
- response = self.c.post(reverse('oozie:workflow_validate_action', kwargs={'workflow': self.wf.pk, 'node_type': 'hive'}), data={'node': json.dumps(data)}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(-1, test_response_json_object['status'])
- assert_true('name' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['name']), test_response_json_object['data'])
- assert_true('description' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['description']), test_response_json_object['data'])
- assert_true('script_path' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['script_path']), test_response_json_object['data'])
- assert_true('job_xml' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['job_xml']), test_response_json_object['data'])
- assert_true('job_properties' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['job_properties']), test_response_json_object['data'])
- assert_true('files' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(1, len(test_response_json_object['data']['files']), test_response_json_object['data'])
- assert_true('params' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['params']), test_response_json_object['data'])
- assert_true('prepares' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['prepares']), test_response_json_object['data'])
- assert_true('archives' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['archives']), test_response_json_object['data'])
- # Empty script path
- data = {"files":"[\"hive-site.xml\"]","job_xml":"hive-site.xml","description":"Show databases","workflow":17,"child_links":[{"comment":"","name":"ok","id":106,"parent":76,"child":74},{"comment":"","name":"error","id":107,"parent":76,"child":73}],"job_properties":"[{\"name\":\"oozie.hive.defaults\",\"value\":\"hive-site.xml\"}]","node_type":"hive","params":"[{\"value\":\"INPUT=/user/hue/oozie/workspaces/data\",\"type\":\"param\"}]","archives":"[]","node_ptr":76,"prepares":"[]","script_path":"","id":76,"name":"Hive"}
- response = self.c.post(reverse('oozie:workflow_validate_action', kwargs={'workflow': self.wf.pk, 'node_type': 'hive'}), data={'node': json.dumps(data)}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(-1, test_response_json_object['status'])
- assert_true('name' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['name']), test_response_json_object['data'])
- assert_true('description' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['description']), test_response_json_object['data'])
- assert_true('script_path' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(1, len(test_response_json_object['data']['script_path']), test_response_json_object['data'])
- assert_true('job_xml' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['job_xml']), test_response_json_object['data'])
- assert_true('job_properties' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['job_properties']), test_response_json_object['data'])
- assert_true('files' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['files']), test_response_json_object['data'])
- assert_true('params' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['params']), test_response_json_object['data'])
- assert_true('prepares' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['prepares']), test_response_json_object['data'])
- assert_true('archives' in test_response_json_object['data'], test_response_json_object['data'])
- assert_equal(0, len(test_response_json_object['data']['archives']), test_response_json_object['data'])
- class TestApiPermissionsWithOozie(OozieBase):
- def setUp(self):
- OozieBase.setUp(self)
- # When updating wf, update wf_json as well!
- self.wf = Workflow.objects.get(name='MapReduce').clone(self.cluster.fs, self.user)
- def test_workflow_save(self):
- # Share
- self.wf.is_shared = True
- self.wf.save()
- Workflow.objects.check_workspace(self.wf, self.cluster.fs)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test')
- grant_access("not_me", "test", "oozie")
- workflow_dict = workflow_to_dict(self.wf)
- workflow_dict = remove_related_fields(workflow_dict)
- workflow_json = json.dumps(workflow_dict)
- response = client_not_me.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- assert_equal(401, response.status_code, response.status_code)
- response = self.c.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(200, response.status_code, response)
- assert_equal(0, test_response_json_object['status'])
- def test_workflow_save_fail(self):
- # Unshare
- self.wf.is_shared = False
- self.wf.save()
- Workflow.objects.check_workspace(self.wf, self.cluster.fs)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test')
- grant_access("not_me", "test", "oozie")
- workflow_dict = workflow_to_dict(self.wf)
- del workflow_dict['name']
- workflow_dict = remove_related_fields( workflow_dict )
- workflow_json = json.dumps(workflow_dict)
- response = client_not_me.post(reverse('oozie:workflow_save', kwargs={'workflow': self.wf.pk}), data={'workflow': workflow_json}, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- assert_equal(401, response.status_code, response)
- def test_workflow(self):
- # Share
- self.wf.is_shared = True
- self.wf.save()
- Workflow.objects.check_workspace(self.wf, self.cluster.fs)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test')
- grant_access("not_me", "test", "oozie")
- response = client_not_me.get(reverse('oozie:workflow', kwargs={'workflow': self.wf.pk}), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- test_response_json = response.content
- test_response_json_object = json.loads(test_response_json)
- assert_equal(200, response.status_code, response)
- assert_equal(0, test_response_json_object['status'])
- def test_workflow_fail(self):
- # Unshare
- self.wf.is_shared = False
- self.wf.save()
- Workflow.objects.check_workspace(self.wf, self.cluster.fs)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test')
- grant_access("not_me", "test", "oozie")
- response = client_not_me.get(reverse('oozie:workflow', kwargs={'workflow': self.wf.pk}), HTTP_X_REQUESTED_WITH='XMLHttpRequest')
- assert_equal(401, response.status_code)
- class TestEditor(OozieMockBase):
- def setUp(self):
- super(TestEditor, self).setUp()
- self.setup_simple_workflow()
- def test_find_parameters(self):
- jobs = [Job(name="$a"),
- Job(name="foo ${b} $$"),
- Job(name="${foo}", description="xxx ${foo}")]
- result = [find_parameters(job, ['name', 'description']) for job in jobs]
- assert_equal(set(["b", "foo"]), reduce(lambda x, y: x | set(y), result, set()))
- def test_find_all_parameters(self):
- assert_equal([{'name': u'output', 'value': u''}, {'name': u'SLEEP', 'value': ''}, {'name': u'market', 'value': u'US'}],
- self.wf.find_all_parameters())
- def test_workflow_has_cycle(self):
- action1 = Node.objects.get(name='action-name-1')
- action3 = Node.objects.get(name='action-name-3')
- assert_false(self.wf.has_cycle())
- ok = action3.get_link('ok')
- ok.child = action1
- ok.save()
- assert_true(self.wf.has_cycle())
- def test_workflow_gen_xml(self):
- assert_equal(
- '<workflow-app name="wf-name-1" xmlns="uri:oozie:workflow:0.2">\n'
- ' <global>\n'
- ' <job-xml>jobconf.xml</job-xml>\n'
- ' <configuration>\n'
- ' <property>\n'
- ' <name>sleep-all</name>\n'
- ' <value>${SLEEP}</value>\n'
- ' </property>\n'
- ' </configuration>\n'
- ' </global>\n'
- ' <start to="action-name-1"/>\n'
- ' <action name="action-name-1">\n'
- ' <map-reduce>\n'
- ' <job-tracker>${jobTracker}</job-tracker>\n'
- ' <name-node>${nameNode}</name-node>\n'
- ' <prepare>\n'
- ' <delete path="${nameNode}${output}"/>\n'
- ' <mkdir path="${nameNode}/test"/>\n'
- ' </prepare>\n'
- ' <configuration>\n'
- ' <property>\n'
- ' <name>sleep</name>\n'
- ' <value>${SLEEP}</value>\n'
- ' </property>\n'
- ' </configuration>\n'
- ' </map-reduce>\n'
- ' <ok to="action-name-2"/>\n'
- ' <error to="kill"/>\n'
- ' </action>\n'
- ' <action name="action-name-2">\n'
- ' <map-reduce>\n'
- ' <job-tracker>${jobTracker}</job-tracker>\n'
- ' <name-node>${nameNode}</name-node>\n'
- ' <prepare>\n'
- ' <delete path="${nameNode}${output}"/>\n'
- ' <mkdir path="${nameNode}/test"/>\n'
- ' </prepare>\n'
- ' <configuration>\n'
- ' <property>\n'
- ' <name>sleep</name>\n'
- ' <value>${SLEEP}</value>\n'
- ' </property>\n'
- ' </configuration>\n'
- ' </map-reduce>\n'
- ' <ok to="action-name-3"/>\n'
- ' <error to="kill"/>\n'
- ' </action>\n'
- ' <action name="action-name-3">\n'
- ' <map-reduce>\n'
- ' <job-tracker>${jobTracker}</job-tracker>\n'
- ' <name-node>${nameNode}</name-node>\n'
- ' <prepare>\n'
- ' <delete path="${nameNode}${output}"/>\n'
- ' <mkdir path="${nameNode}/test"/>\n'
- ' </prepare>\n'
- ' <configuration>\n'
- ' <property>\n'
- ' <name>sleep</name>\n'
- ' <value>${SLEEP}</value>\n'
- ' </property>\n'
- ' </configuration>\n'
- ' </map-reduce>\n'
- ' <ok to="end"/>\n'
- ' <error to="kill"/>\n'
- ' </action>\n'
- ' <kill name="kill">\n'
- ' <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>\n'
- ' </kill>\n'
- ' <end name="end"/>\n'
- '</workflow-app>'.split(), self.wf.to_xml().split())
- def test_workflow_shell_gen_xml(self):
- self.wf.node_set.filter(name='action-name-1').delete()
- action1 = add_node(self.wf, 'action-name-1', 'shell', [self.wf.start], {
- u'job_xml': 'my-job.xml',
- u'files': '["hello.py"]',
- u'name': 'Shell',
- u'job_properties': '[]',
- u'capture_output': 'on',
- u'command': 'hello.py',
- u'archives': '[]',
- u'prepares': '[]',
- u'params': '[{"value":"World!","type":"argument"}]',
- u'description': 'Execute a Python script printing its arguments'
- })
- Link(parent=action1, child=self.wf.end, name="ok").save()
- xml = self.wf.to_xml()
- assert_true("""
- <shell xmlns="uri:oozie:shell-action:0.1">
- <job-tracker>${jobTracker}</job-tracker>
- <name-node>${nameNode}</name-node>
- <job-xml>my-job.xml</job-xml>
- <exec>hello.py</exec>
- <argument>World!</argument>
- <file>hello.py#hello.py</file>
- <capture-output/>
- </shell>""" in xml, xml)
- action1.capture_output = False
- action1.save()
- xml = self.wf.to_xml()
- assert_true("""
- <shell xmlns="uri:oozie:shell-action:0.1">
- <job-tracker>${jobTracker}</job-tracker>
- <name-node>${nameNode}</name-node>
- <job-xml>my-job.xml</job-xml>
- <exec>hello.py</exec>
- <argument>World!</argument>
- <file>hello.py#hello.py</file>
- </shell>""" in xml, xml)
- def test_workflow_fs_gen_xml(self):
- self.wf.node_set.filter(name='action-name-1').delete()
- action1 = add_node(self.wf, 'action-name-1', 'fs', [self.wf.start], {
- u'name': 'MyFs',
- u'description': 'Execute a Fs action that manage files',
- u'deletes': '[{"name":"/to/delete"},{"name":"to/delete2"}]',
- u'mkdirs': '[{"name":"/to/mkdir"},{"name":"${mkdir2}"}]',
- u'moves': '[{"source":"/to/move/source","destination":"/to/move/destination"},{"source":"/to/move/source2","destination":"/to/move/destination2"}]',
- u'chmods': '[{"path":"/to/chmod","recursive":true,"permissions":"-rwxrw-rw-"},{"path":"/to/chmod2","recursive":false,"permissions":"755"}]',
- u'touchzs': '[{"name":"/to/touchz"},{"name":"/to/touchz2"}]'
- })
- Link(parent=action1, child=self.wf.end, name="ok").save()
- xml = self.wf.to_xml()
- assert_true("""
- <action name="MyFs">
- <fs>
- <delete path='${nameNode}/to/delete'/>
- <delete path='${nameNode}/user/${wf:user()}/to/delete2'/>
- <mkdir path='${nameNode}/to/mkdir'/>
- <mkdir path='${nameNode}${mkdir2}'/>
- <move source='${nameNode}/to/move/source' target='${nameNode}/to/move/destination'/>
- <move source='${nameNode}/to/move/source2' target='${nameNode}/to/move/destination2'/>
- <chmod path='${nameNode}/to/chmod' permissions='-rwxrw-rw-' dir-files='true'/>
- <chmod path='${nameNode}/to/chmod2' permissions='755' dir-files='false'/>
- <touchz path='${nameNode}/to/touchz'/>
- <touchz path='${nameNode}/to/touchz2'/>
- </fs>
- <ok to="end"/>
- <error to="kill"/>
- </action>""" in xml, xml)
- def test_workflow_email_gen_xml(self):
- self.wf.node_set.filter(name='action-name-1').delete()
- action1 = add_node(self.wf, 'action-name-1', 'email', [self.wf.start], {
- u'name': 'MyEmail',
- u'description': 'Execute an Email action',
- u'to': 'hue@hue.org,django@python.org',
- u'cc': '',
- u'subject': 'My subject',
- u'body': 'My body'
- })
- Link(parent=action1, child=self.wf.end, name="ok").save()
- xml = self.wf.to_xml()
- assert_true("""
- <action name="MyEmail">
- <email xmlns="uri:oozie:email-action:0.1">
- <to>hue@hue.org,django@python.org</to>
- <subject>My subject</subject>
- <body>My body</body>
- </email>
- <ok to="end"/>
- <error to="kill"/>
- </action>""" in xml, xml)
- action1.cc = 'lambda@python.org'
- action1.save()
- xml = self.wf.to_xml()
- assert_true("""
- <action name="MyEmail">
- <email xmlns="uri:oozie:email-action:0.1">
- <to>hue@hue.org,django@python.org</to>
- <cc>lambda@python.org</cc>
- <subject>My subject</subject>
- <body>My body</body>
- </email>
- <ok to="end"/>
- <error to="kill"/>
- </action>""" in xml, xml)
- def test_workflow_subworkflow_gen_xml(self):
- self.wf.node_set.filter(name='action-name-1').delete()
- wf_dict = WORKFLOW_DICT.copy()
- wf_dict['name'] = [u'wf-name-2']
- wf2 = create_workflow(self.c, wf_dict)
- action1 = add_node(self.wf, 'action-name-1', 'subworkflow', [self.wf.start], {
- u'name': 'MySubworkflow',
- u'description': 'Execute a subworkflow action',
- u'sub_workflow': wf2,
- u'propagate_configuration': True,
- u'job_properties': '[{"value":"World!","name":"argument"}]'
- })
- Link(parent=action1, child=self.wf.end, name="ok").save()
- xml = self.wf.to_xml()
- assert_true(re.search(
- '<sub-workflow>\W+'
- '<app-path>\${nameNode}/user/hue/oozie/workspaces/_test_-oozie-(.+?)</app-path>\W+'
- '<propagate-configuration/>\W+'
- '<configuration>\W+'
- '<property>\W+'
- '<name>argument</name>\W+'
- '<value>World!</value>\W+'
- '</property>\W+'
- '</configuration>\W+'
- '</sub-workflow>', xml, re.MULTILINE), xml)
- wf2.delete()
- def test_workflow_flatten_list(self):
- assert_equal('[<Start: start>, <Mapreduce: action-name-1>, <Mapreduce: action-name-2>, <Mapreduce: action-name-3>, '
- '<Kill: kill>, <End: end>]',
- str(self.wf.node_list))
- # 1 2
- # 3
- self.setup_forking_workflow()
- assert_equal('[<Start: start>, <Fork: fork-name-1>, <Mapreduce: action-name-1>, <Mapreduce: action-name-2>, '
- '<Join: join-name-1>, <Mapreduce: action-name-3>, <Kill: kill>, <End: end>]',
- str(self.wf.node_list))
- def test_create_coordinator(self):
- create_coordinator(self.wf, self.c)
- def test_clone_coordinator(self):
- coord = create_coordinator(self.wf, self.c)
- coordinator_count = Coordinator.objects.count()
- response = self.c.post(reverse('oozie:clone_coordinator', args=[coord.id]), {}, follow=True)
- coord2 = Coordinator.objects.latest('id')
- assert_not_equal(coord.id, coord2.id)
- assert_equal(coordinator_count + 1, Coordinator.objects.count(), response)
- assert_equal(coord.dataset_set.count(), coord2.dataset_set.count())
- assert_equal(coord.datainput_set.count(), coord2.datainput_set.count())
- assert_equal(coord.dataoutput_set.count(), coord2.dataoutput_set.count())
- ds_ids = set(coord.dataset_set.values_list('id', flat=True))
- for node in coord2.dataset_set.all():
- assert_false(node.id in ds_ids)
- data_input_ids = set(coord.datainput_set.values_list('id', flat=True))
- for node in coord2.datainput_set.all():
- assert_false(node.id in data_input_ids)
- data_output_ids = set(coord.dataoutput_set.values_list('id', flat=True))
- for node in coord2.dataoutput_set.all():
- assert_false(node.id in data_output_ids)
- assert_not_equal(coord.deployment_dir, coord2.deployment_dir)
- assert_not_equal('', coord2.deployment_dir)
- def test_coordinator_workflow_access_permissions(self):
- self.wf.is_shared = True
- self.wf.save()
- # Login as someone else not superuser
- client_another_me = make_logged_in_client(username='another_me', is_superuser=False, groupname='test')
- grant_access("another_me", "test", "oozie")
- coord = create_coordinator(self.wf, client_another_me)
- response = client_another_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true('Editor' in response.content, response.content)
- assert_true('value="Save"' in response.content, response.content)
- # Check can schedule a non personal/shared workflow
- workflow_select = '%s</option>' % self.wf
- response = client_another_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true(workflow_select in response.content, response.content)
- self.wf.is_shared = False
- self.wf.save()
- response = client_another_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_false(workflow_select in response.content, response.content)
- self.wf.is_shared = True
- self.wf.save()
- # Edit
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_another_me.post(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true(workflow_select in response.content, response.content)
- assert_true('value="Save"' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_another_me.post(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true('This field is required' in response.content, response.content)
- assert_false(workflow_select in response.content, response.content)
- assert_true('value="Save"' in response.content, response.content)
- finally:
- finish()
- def test_coordinator_gen_xml(self):
- coord = create_coordinator(self.wf, self.c)
- assert_equal(
- '<coordinator-app name="MyCoord"\n'
- ' frequency="${coord:days(1)}"\n'
- ' start="2012-07-01T00:00Z" end="2012-07-04T00:00Z" timezone="America/Los_Angeles"\n'
- ' xmlns="uri:oozie:coordinator:0.1">\n'
- ' <controls>\n'
- ' <timeout>100</timeout>\n'
- ' <concurrency>3</concurrency>\n'
- ' <execution>FIFO</execution>\n'
- ' <throttle>10</throttle>\n'
- ' </controls>\n'
- ' <action>\n'
- ' <workflow>\n'
- ' <app-path>${wf_application_path}</app-path>\n'
- ' </workflow>\n'
- ' </action>\n'
- '</coordinator-app>\n'.split(), coord.to_xml().split())
- def test_coordinator_with_data_input_gen_xml(self):
- coord = create_coordinator(self.wf, self.c)
- create_dataset(coord, self.c)
- create_coordinator_data(coord, self.c)
- assert_equal(
- ['<coordinator-app', 'name="MyCoord"', 'frequency="${coord:days(1)}"', 'start="2012-07-01T00:00Z"', 'end="2012-07-04T00:00Z"',
- 'timezone="America/Los_Angeles"',
- 'xmlns="uri:oozie:coordinator:0.1">',
- '<controls>',
- '<timeout>100</timeout>',
- '<concurrency>3</concurrency>',
- '<execution>FIFO</execution>',
- '<throttle>10</throttle>',
- '</controls>',
- '<datasets>',
- '<dataset', 'name="MyDataset"', 'frequency="${coord:days(1)}"', 'initial-instance="2012-07-01T00:00Z"', 'timezone="America/Los_Angeles">',
- '<uri-template>/data/${YEAR}${MONTH}${DAY}</uri-template>',
- '<done-flag></done-flag>',
- '</dataset>',
- '</datasets>',
- '<input-events>',
- '<data-in', 'name="input_dir"', 'dataset="MyDataset">',
- '<instance>${coord:current(0)}</instance>',
- '</data-in>',
- '</input-events>',
- '<action>',
- '<workflow>',
- '<app-path>${wf_application_path}</app-path>',
- '<configuration>',
- '<property>',
- '<name>input_dir</name>',
- "<value>${coord:dataIn('input_dir')}</value>",
- '</property>',
- '</configuration>',
- '</workflow>',
- '</action>',
- '</coordinator-app>'], coord.to_xml().split())
- def test_create_coordinator_dataset(self):
- coord = create_coordinator(self.wf, self.c)
- create_dataset(coord, self.c)
- def test_create_coordinator_input_data(self):
- coord = create_coordinator(self.wf, self.c)
- create_dataset(coord, self.c)
- create_coordinator_data(coord, self.c)
- def test_setup_app(self):
- self.c.post(reverse('oozie:setup_app'))
- def test_workflow_prepare(self):
- action1 = Node.objects.get(name='action-name-1').get_full_node()
- action1.prepares = json.dumps([
- {"type": "delete","value": "${output}"},
- {"type": "delete","value": "out"},
- {"type": "delete","value": "/user/test/out"},
- {"type": "delete","value": "hdfs://localhost:8020/user/test/out"}])
- action1.save()
- xml = self.wf.to_xml()
- assert_true('<delete path="${nameNode}${output}"/>' in xml, xml)
- assert_true('<delete path="${nameNode}/user/${wf:user()}/out"/>' in xml, xml)
- assert_true('<delete path="${nameNode}/user/test/out"/>' in xml, xml)
- assert_true('<delete path="hdfs://localhost:8020/user/test/out"/>' in xml, xml)
- def test_get_workflow_parameters(self):
- assert_equal([{'name': u'output', 'value': ''}, {'name': u'SLEEP', 'value': ''}, {'name': u'market', 'value': u'US'}],
- self.wf.find_all_parameters())
- def test_get_coordinator_parameters(self):
- coord = create_coordinator(self.wf, self.c)
- create_dataset(coord, self.c)
- create_coordinator_data(coord, self.c)
- assert_equal([{'name': u'output', 'value': ''}, {'name': u'SLEEP', 'value': ''}, {'name': u'market', 'value': u'US,France'}],
- coord.find_all_parameters())
- def test_workflow_data_binds(self):
- response = self.c.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
- assert_equal(1, response.content.count('checked: is_shared'), response.content)
- assert_true('checked: capture_output' in response.content, response.content)
- def test_import_workflow_basic(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-basic.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(2, len(Node.objects.filter(workflow=workflow)))
- assert_equal(1, len(Link.objects.filter(parent__workflow=workflow)))
- assert_equal('done', Node.objects.get(workflow=workflow, node_type='end').name)
- workflow.delete()
- def test_import_workflow_decision(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-decision.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(11, len(Node.objects.filter(workflow=workflow)))
- assert_equal(19, len(Link.objects.filter(parent__workflow=workflow)))
- assert_equal(1, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='decision', comment='${1 gt 2}', name='start')))
- assert_equal(1, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='decision', comment='', name='start')))
- assert_equal(1, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='decision', name='default')))
- assert_equal(1, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='decision', child__node_type='end', name='related')))
- workflow.delete()
- def test_import_workflow_distcp(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-distcp.0.1.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(4, len(Node.objects.filter(workflow=workflow)))
- assert_equal(3, len(Link.objects.filter(parent__workflow=workflow)))
- assert_equal('[{"type":"arg","value":"-overwrite"},{"type":"arg","value":"-m"},{"type":"arg","value":"${MAP_NUMBER}"},{"type":"arg","value":"/user/hue/oozie/workspaces/data"},{"type":"arg","value":"${OUTPUT}"}]', Node.objects.get(workflow=workflow, node_type='distcp').get_full_node().params)
- workflow.delete()
- def test_import_workflow_forks(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-forks.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(12, len(Node.objects.filter(workflow=workflow)))
- assert_equal(19, len(Link.objects.filter(parent__workflow=workflow)))
- assert_equal(6, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='fork')))
- assert_equal(4, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='fork', name='start')))
- assert_equal(2, len(Link.objects.filter(parent__workflow=workflow, parent__node_type='fork', child__node_type='join', name='related')))
- workflow.delete()
- def test_import_workflow_mapreduce(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-mapreduce.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(4, len(Node.objects.filter(workflow=workflow)))
- assert_equal(3, len(Link.objects.filter(parent__workflow=workflow)))
- assert_equal('[{"name":"mapred.reduce.tasks","value":"1"},{"name":"mapred.mapper.class","value":"org.apache.hadoop.examples.SleepJob"},{"name":"mapred.reducer.class","value":"org.apache.hadoop.examples.SleepJob"},{"name":"mapred.mapoutput.key.class","value":"org.apache.hadoop.io.IntWritable"},{"name":"mapred.mapoutput.value.class","value":"org.apache.hadoop.io.NullWritable"},{"name":"mapred.output.format.class","value":"org.apache.hadoop.mapred.lib.NullOutputFormat"},{"name":"mapred.input.format.class","value":"org.apache.hadoop.examples.SleepJob$SleepInputFormat"},{"name":"mapred.partitioner.class","value":"org.apache.hadoop.examples.SleepJob"},{"name":"mapred.speculative.execution","value":"false"},{"name":"sleep.job.map.sleep.time","value":"0"},{"name":"sleep.job.reduce.sleep.time","value":"1"}]', Node.objects.get(workflow=workflow, node_type='mapreduce').get_full_node().job_properties)
- workflow.delete()
- def test_import_workflow_pig(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-pig.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- node = Node.objects.get(workflow=workflow, node_type='pig').get_full_node()
- assert_equal(4, len(Node.objects.filter(workflow=workflow)))
- assert_equal(3, len(Link.objects.filter(parent__workflow=workflow)))
- assert_equal('aggregate.pig', node.script_path)
- assert_equal('[{"type":"argument","value":"-param"},{"type":"argument","value":"INPUT=/user/hue/oozie/workspaces/data"},{"type":"argument","value":"-param"},{"type":"argument","value":"OUTPUT=${output}"}]', node.params)
- workflow.delete()
- def test_import_workflow_sqoop(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-sqoop.0.2.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(4, len(Node.objects.filter(workflow=workflow)))
- assert_equal(3, len(Link.objects.filter(parent__workflow=workflow)))
- node = Node.objects.get(workflow=workflow, node_type='sqoop').get_full_node()
- assert_equal('["db.hsqldb.properties#db.hsqldb.properties","db.hsqldb.script#db.hsqldb.script"]', node.files)
- assert_equal('import --connect jdbc:hsqldb:file:db.hsqldb --table TT --target-dir ${output} -m 1', node.script_path)
- workflow.delete()
- def test_import_workflow_java(self):
- workflow = Workflow.objects.new_workflow(self.user)
- workflow.save()
- f = open('apps/oozie/src/oozie/test_data/0.4/test-java.xml')
- import_workflow(workflow, f.read(), schema_version=0.4)
- f.close()
- workflow.save()
- assert_equal(5, len(Node.objects.filter(workflow=workflow)))
- assert_equal(5, len(Link.objects.filter(parent__workflow=workflow)))
- nodes = [Node.objects.filter(workflow=workflow, node_type='java')[0].get_full_node(),
- Node.objects.filter(workflow=workflow, node_type='java')[1].get_full_node()]
- assert_equal('org.apache.hadoop.examples.terasort.TeraGen', nodes[0].main_class)
- assert_equal('["${records}","${output_dir}/teragen"]', nodes[0].args)
- assert_equal('org.apache.hadoop.examples.terasort.TeraSort', nodes[1].main_class)
- assert_equal('["${output_dir}/teragen","${output_dir}/terasort"]', nodes[1].args)
- workflow.delete()
- class TestPermissions(OozieBase):
- def setUp(self):
- self.c = make_logged_in_client()
- self.wf = create_workflow(self.c)
- self.setup_simple_workflow()
- def test_workflow_permissions(self):
- response = self.c.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
- assert_true('Editor' in response.content, response.content)
- assert_true('Save' in response.content, response.content)
- assert_false(self.wf.is_shared)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test')
- grant_access("not_me", "test", "oozie")
- # List
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:list_workflows'))
- assert_false('wf-name-1' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.get(reverse('oozie:list_workflows'))
- assert_false('wf-name-1' in response.content, response.content)
- finally:
- finish()
- # View
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- # Share it !
- self.wf = Workflow.objects.get(name='wf-name-1')
- self.wf.is_shared = True
- self.wf.save()
- Workflow.objects.check_workspace(self.wf, self.cluster.fs)
- # List
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:list_workflows'))
- assert_equal(200, response.status_code)
- assert_true('wf-name-1' in response.content, response.content)
- finally:
- finish()
- # View
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
- assert_false('Permission denied' in response.content, response.content)
- assert_true('Save' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- # Submit
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.post(reverse('oozie:submit_workflow', args=[self.wf.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- try:
- response = client_not_me.post(reverse('oozie:submit_workflow', args=[self.wf.id]))
- assert_false('Permission denied' in response.content, response.content)
- except IOError:
- pass
- finally:
- finish()
- # Delete
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.post(reverse('oozie:delete_workflow', args=[self.wf.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- response = self.c.post(reverse('oozie:delete_workflow', args=[self.wf.id]), follow=True)
- assert_equal(200, response.status_code)
- def test_coordinator_permissions(self):
- coord = create_coordinator(self.wf, self.c)
- response = self.c.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true('Editor' in response.content, response.content)
- assert_true('value="Save"' in response.content, response.content)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test')
- grant_access("not_me", "test", "oozie")
- # List
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:list_coordinators'))
- assert_false('MyCoord' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.get(reverse('oozie:list_coordinators'))
- assert_false('MyCoord' in response.content, response.content)
- finally:
- finish()
- # View
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_false('MyCoord' in response.content, response.content)
- finally:
- finish()
- # Share it !
- wf = Workflow.objects.get(id=coord.workflow.id)
- wf.is_shared = True
- wf.save()
- Workflow.objects.check_workspace(wf, self.cluster.fs)
- post = COORDINATOR_DICT.copy()
- post.update({
- u'datainput_set-TOTAL_FORMS': [u'0'], u'datainput_set-INITIAL_FORMS': [u'0'], u'dataset_set-INITIAL_FORMS': [u'0'],
- u'dataoutput_set-INITIAL_FORMS': [u'0'], u'datainput_set-MAX_NUM_FORMS': [u'0'], u'output-MAX_NUM_FORMS': [u''],
- u'output-INITIAL_FORMS': [u'0'], u'dataoutput_set-TOTAL_FORMS': [u'0'], u'input-TOTAL_FORMS': [u'0'],
- u'dataset_set-MAX_NUM_FORMS': [u'0'], u'dataoutput_set-MAX_NUM_FORMS': [u'0'], u'input-MAX_NUM_FORMS': [u''],
- u'dataset_set-TOTAL_FORMS': [u'0'], u'input-INITIAL_FORMS': [u'0'], u'output-TOTAL_FORMS': [u'0']})
- post['is_shared'] = [u'on']
- post['workflow'] = coord.workflow.id
- self.c.post(reverse('oozie:edit_coordinator', args=[coord.id]), post)
- coord = Coordinator.objects.get(id=coord.id)
- assert_true(coord.is_shared)
- # List
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:list_coordinators'))
- assert_equal(200, response.status_code)
- assert_true('MyCoord' in response.content, response.content)
- finally:
- finish()
- # View
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_false('Permission denied' in response.content, response.content)
- assert_false('value="Save"' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.get(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- # Edit
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- response = client_not_me.post(reverse('oozie:edit_coordinator', args=[coord.id]))
- assert_false('MyCoord' in response.content, response.content)
- assert_true('Not allowed' in response.content, response.content)
- finally:
- finish()
- # Submit
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.post(reverse('oozie:submit_coordinator', args=[coord.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- try:
- response = client_not_me.post(reverse('oozie:submit_coordinator', args=[coord.id]))
- assert_false('Permission denied' in response.content, response.content)
- except IOError:
- pass
- finally:
- finish()
- # Resubmit
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- history, created = History.objects.get_or_create(job=coord, oozie_job_id=MockOozieApi.COORDINATOR_IDS[0],
- defaults={'submitter': User.objects.get(username='test'), 'properties': '[]'})
- oozie_job_id = history.oozie_job_id
- response = client_not_me.post(reverse('oozie:resubmit_coordinator', args=[oozie_job_id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- finish = SHARE_JOBS.set_for_testing(True)
- try:
- try:
- history, created = History.objects.get_or_create(job=coord, oozie_job_id=MockOozieApi.COORDINATOR_IDS[0],
- defaults={'submitter': User.objects.get(username='test'), 'properties': '[]'})
- oozie_job_id = history.oozie_job_id
- response = client_not_me.post(reverse('oozie:resubmit_coordinator', args=[oozie_job_id]))
- assert_false('Permission denied' in response.content, response.content)
- except IOError:
- pass
- finally:
- finish()
- # Delete
- finish = SHARE_JOBS.set_for_testing(False)
- try:
- response = client_not_me.post(reverse('oozie:delete_coordinator', args=[coord.id]))
- assert_true('Permission denied' in response.content, response.content)
- finally:
- finish()
- response = self.c.post(reverse('oozie:delete_coordinator', args=[coord.id]), follow=True)
- assert_equal(200, response.status_code)
- class TestEditorWithOozie(OozieBase):
- def setUp(self):
- self.c = make_logged_in_client()
- self.wf = create_workflow(self.c)
- self.setup_simple_workflow()
- def tearDown(self):
- self.wf.delete()
- def test_create_workflow(self):
- dir_stat = self.cluster.fs.stats(self.wf.deployment_dir)
- assert_equal('test', dir_stat.user)
- assert_equal('hue', dir_stat.group)
- assert_equal('40711', '%o' % dir_stat.mode)
- def test_clone_workflow(self):
- workflow_count = Workflow.objects.count()
- response = self.c.post(reverse('oozie:clone_workflow', args=[self.wf.id]), {}, follow=True)
- assert_equal(workflow_count + 1, Workflow.objects.count(), response)
- wf2 = Workflow.objects.latest('id')
- assert_not_equal(self.wf.id, wf2.id)
- assert_equal(self.wf.node_set.count(), wf2.node_set.count())
- node_ids = set(self.wf.node_set.values_list('id', flat=True))
- for node in wf2.node_set.all():
- assert_false(node.id in node_ids)
- assert_not_equal(self.wf.deployment_dir, wf2.deployment_dir)
- assert_not_equal('', wf2.deployment_dir)
- def test_import_action(self):
- raise SkipTest
- # Setup jobsub examples
- if not jobsub_setup.Command().has_been_setup():
- jobsub_setup.Command().handle()
- # There should be 3 from examples
- jobsub_design = OozieDesign.objects.all()[0]
- node_size = len(Node.objects.all())
- kwargs = dict(workflow=self.wf.id, parent_action_id=self.wf.end.get_parents()[0].id)
- response = self.c.post(reverse('oozie:import_action', kwargs=kwargs), {'action_id': jobsub_design.id})
- assert_equal(302, response.status_code)
- assert_equal(node_size + 1, len(Node.objects.all()))
- # There should now be an imported action at the end of Node list
- # Need to test properties to make sure we got it right
- # Must also make sure that jobsub field values are translated
- translation_regex = re.compile('(?<!\$)\$(\w+)')
- node = Node.objects.all()[len(Node.objects.all())-1].get_full_node()
- for field in node.PARAM_FIELDS:
- assert_equal(translation_regex.sub(r'${\1}', getattr(jobsub_design.get_root_action(), field)), getattr(node, field))
- def test_import_workflow(self):
- workflow_count = Workflow.objects.count()
- # Create
- filename = os.path.abspath(os.path.dirname(__file__) + "/test_data/0.4/test-mapreduce.xml")
- fh = open(filename)
- response = self.c.post(reverse('oozie:import_workflow'), {
- 'job_xml': [''],
- 'name': ['test_workflow'],
- 'parameters': ['[{"name":"oozie.use.system.libpath","value":"true"}]'],
- 'deployment_dir': [''],
- 'job_properties': ['[]'],
- 'schema_version': ['0.4'],
- 'definition_file': [fh],
- 'description': ['']
- }, follow=True)
- fh.close()
- assert_equal(workflow_count + 1, Workflow.objects.count(), response)
- class TestOozieSubmissions(OozieBase):
- def test_submit_mapreduce_action(self):
- wf = Workflow.objects.get(name='MapReduce')
- post_data = {u'form-MAX_NUM_FORMS': [u''], u'form-INITIAL_FORMS': [u'1'],
- u'form-0-name': [u'REDUCER_SLEEP_TIME'], u'form-0-value': [u'1'], u'form-TOTAL_FORMS': [u'1']}
- response = self.c.post(reverse('oozie:submit_workflow', args=[wf.id]), data=post_data, follow=True)
- job = OozieServerProvider.wait_until_completion(response.context['oozie_workflow'].id)
- assert_equal('SUCCEEDED', job.status)
- # Rerun with default options
- post_data.update({u'rerun_form_choice': [u'skip_nodes']})
- response = self.c.post(reverse('oozie:rerun_oozie_job', kwargs={'job_id': job.id, 'app_path': job.appPath}), data=post_data, follow=True)
- job = OozieServerProvider.wait_until_completion(response.context['oozie_workflow'].id)
- assert_equal('SUCCEEDED', job.status)
- # Rerun with skip OK actions skipped
- post_data.update({u'rerun_form_choice': [u'skip_nodes'], u'skip_nodes': [u'Sleep']})
- response = self.c.post(reverse('oozie:rerun_oozie_job', kwargs={'job_id': job.id, 'app_path': job.appPath}), data=post_data, follow=True)
- job = OozieServerProvider.wait_until_completion(response.context['oozie_workflow'].id)
- assert_equal('SUCCEEDED', job.status)
- # Rerun with failed nodes too
- post_data.update({u'rerun_form_choice': [u'failed_nodes']})
- response = self.c.post(reverse('oozie:rerun_oozie_job', kwargs={'job_id': job.id, 'app_path': job.appPath}), data=post_data, follow=True)
- job = OozieServerProvider.wait_until_completion(response.context['oozie_workflow'].id)
- def test_submit_java_action(self):
- wf = Workflow.objects.get(name='Sequential Java')
- response = self.c.post(reverse('oozie:submit_workflow', args=[wf.id]),
- data={u'form-MAX_NUM_FORMS': [u''],
- u'form-0-name': [u'records'], u'form-0-value': [u'10'],
- u'form-1-name': [u' output_dir '], u'form-1-value': [u'${nameNode}/user/test/out/terasort'],
- u'form-INITIAL_FORMS': [u'2'], u'form-TOTAL_FORMS': [u'2']},
- follow=True)
- job = OozieServerProvider.wait_until_completion(response.context['oozie_workflow'].id)
- assert_equal('SUCCEEDED', job.status)
- def test_submit_distcp_action(self):
- wf = Workflow.objects.get(name='DistCp')
- response = self.c.post(reverse('oozie:submit_workflow', args=[wf.id]),
- data= {u'form-MAX_NUM_FORMS': [u''], u'form-TOTAL_FORMS': [u'3'], u'form-INITIAL_FORMS': [u'3'],
- u'form-0-name': [u'oozie.use.system.libpath'], u'form-0-value': [u'true'],
- u'form-1-name': [u'OUTPUT'], u'form-1-value': [u'${nameNode}/user/test/out/distcp'],
- u'form-2-name': [u'MAP_NUMBER'], u'form-2-value': [u'5'],
- },
- follow=True)
- job = OozieServerProvider.wait_until_completion(response.context['oozie_workflow'].id)
- assert_equal('SUCCEEDED', job.status)
- class TestDashboardNoMocking:
- def test_oozie_not_running_message(self):
- c = make_logged_in_client(is_superuser=False)
- grant_access("test", "test", "oozie")
- finish = OOZIE_URL.set_for_testing('http://not_localhost:11000/bad')
- try:
- response = c.get(reverse('oozie:list_oozie_workflows'))
- assert_true('The Oozie server is not running' in response.content, response.content)
- finally:
- finish()
- class TestDashboard(OozieMockBase):
- def test_manage_workflow_dashboard(self):
- # Kill button in response
- response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0]]), {}, follow=True)
- assert_true(('%s/kill' % MockOozieApi.WORKFLOW_IDS[0]) in response.content, response.content)
- assert_false('Rerun' in response.content, response.content)
- # Rerun button in response
- response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[1]]), {}, follow=True)
- assert_false(('%s/kill' % MockOozieApi.WORKFLOW_IDS[1]) in response.content, response.content)
- assert_true('Rerun' in response.content, response.content)
- def test_manage_coordinator_dashboard(self):
- # Kill button in response
- response = self.c.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[0]]), {}, follow=True)
- assert_true(('%s/kill' % MockOozieApi.COORDINATOR_IDS[0]) in response.content, response.content)
- assert_false('Rerun' in response.content, response.content)
- # Rerun button in response
- response = self.c.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[1]]), {}, follow=True)
- assert_false(('%s/kill' % MockOozieApi.COORDINATOR_IDS[1]) in response.content, response.content)
- assert_true('Resubmit' in response.content, response.content)
- def test_list_workflows(self):
- response = self.c.get(reverse('oozie:list_oozie_workflows'))
- for wf_id in MockOozieApi.WORKFLOW_IDS:
- assert_true(wf_id in response.content, response.content)
- def test_list_coordinators(self):
- response = self.c.get(reverse('oozie:list_oozie_coordinators'))
- for coord_id in MockOozieApi.COORDINATOR_IDS:
- assert_true(coord_id in response.content, response.content)
- def test_list_workflow(self):
- response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0]]))
- assert_true('Workflow WordCount1' in response.content, response.content)
- assert_true('Workflow' in response.content, response.content)
- response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0], MockOozieApi.COORDINATOR_IDS[0]]))
- assert_true('Workflow WordCount1' in response.content, response.content)
- assert_true('Workflow' in response.content, response.content)
- assert_true('DailyWordCount1' in response.content, response.content)
- assert_true('Coordinator' in response.content, response.content)
- def test_list_coordinator(self):
- response = self.c.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[0]]))
- assert_true('Coordinator DailyWordCount1' in response.content, response.content)
- assert_true('Workflow' in response.content, response.content)
- def test_manage_oozie_jobs(self):
- try:
- self.c.get(reverse('oozie:manage_oozie_jobs', args=[MockOozieApi.COORDINATOR_IDS[0], 'kill']))
- assert False
- except:
- pass
- response = self.c.post(reverse('oozie:manage_oozie_jobs', args=[MockOozieApi.COORDINATOR_IDS[0], 'kill']))
- data = json.loads(response.content)
- assert_equal(0, data['status'])
- def test_workflows_permissions(self):
- response = self.c.get(reverse('oozie:list_oozie_workflows'))
- assert_true('WordCount1' in response.content, response.content)
- # Rerun
- response = self.c.get(reverse('oozie:rerun_oozie_job', kwargs={'job_id': MockOozieApi.WORKFLOW_IDS[0],
- 'app_path': MockOozieApi.JSON_WORKFLOW_LIST[0]['appPath']}))
- assert_false('Permission denied.' in response.content, response.content)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test', recreate=True)
- grant_access("not_me", "not_me", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_workflows'))
- assert_false('WordCount1' in response.content, response.content)
- # Rerun
- response = client_not_me.get(reverse('oozie:rerun_oozie_job', kwargs={'job_id': MockOozieApi.WORKFLOW_IDS[0],
- 'app_path': MockOozieApi.JSON_WORKFLOW_LIST[0]['appPath']}))
- assert_true('Permission denied.' in response.content, response.content)
- # Add read only access
- add_permission("not_me", "dashboard_jobs_access", "dashboard_jobs_access", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_workflows'))
- assert_true('WordCount1' in response.content, response.content)
- # Rerun
- response = client_not_me.get(reverse('oozie:rerun_oozie_job', kwargs={'job_id': MockOozieApi.WORKFLOW_IDS[0],
- 'app_path': MockOozieApi.JSON_WORKFLOW_LIST[0]['appPath']}))
- assert_false('Permission denied.' in response.content, response.content)
- def test_workflow_permissions(self):
- response = self.c.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0]]))
- assert_true('WordCount1' in response.content, response.content)
- assert_false('Permission denied' in response.content, response.content)
- response = self.c.get(reverse('oozie:list_oozie_workflow_action', args=['XXX']))
- assert_false('Permission denied' in response.content, response.content)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test', recreate=True)
- grant_access("not_me", "not_me", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0]]))
- assert_true('Permission denied' in response.content, response.content)
- response = client_not_me.get(reverse('oozie:list_oozie_workflow_action', args=['XXX']))
- assert_true('Permission denied' in response.content, response.content)
- # Add read only access
- add_permission("not_me", "dashboard_jobs_access", "dashboard_jobs_access", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_workflow', args=[MockOozieApi.WORKFLOW_IDS[0]]))
- assert_false('Permission denied' in response.content, response.content)
- def test_coordinators_permissions(self):
- response = self.c.get(reverse('oozie:list_oozie_coordinators'))
- assert_true('DailyWordCount1' in response.content, response.content)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test', recreate=True)
- grant_access("not_me", "not_me", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_coordinators'))
- assert_false('DailyWordCount1' in response.content, response.content)
- # Add read only access
- add_permission("not_me", "dashboard_jobs_access", "dashboard_jobs_access", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_coordinators'))
- assert_true('DailyWordCount1' in response.content, response.content)
- def test_coordinator_permissions(self):
- response = self.c.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[0]]))
- assert_true('DailyWordCount1' in response.content, response.content)
- assert_false('Permission denied' in response.content, response.content)
- # Login as someone else
- client_not_me = make_logged_in_client(username='not_me', is_superuser=False, groupname='test', recreate=True)
- grant_access("not_me", "not_me", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[0]]))
- assert_true('Permission denied' in response.content, response.content)
- # Add read only access
- add_permission("not_me", "dashboard_jobs_access", "dashboard_jobs_access", "oozie")
- response = client_not_me.get(reverse('oozie:list_oozie_coordinator', args=[MockOozieApi.COORDINATOR_IDS[0]]))
- assert_false('Permission denied' in response.content, response.content)
- class TestUtils(OozieMockBase):
- def setUp(self):
- OozieMockBase.setUp(self)
- # When updating wf, update wf_json as well!
- self.wf = Workflow.objects.get(name='wf-name-1')
- def test_workflow_to_dict(self):
- workflow_dict = workflow_to_dict(self.wf)
- # Test properties
- assert_true('job_xml' in workflow_dict, workflow_dict)
- assert_true('is_shared' in workflow_dict, workflow_dict)
- assert_true('end' in workflow_dict, workflow_dict)
- assert_true('description' in workflow_dict, workflow_dict)
- assert_true('parameters' in workflow_dict, workflow_dict)
- assert_true('is_single' in workflow_dict, workflow_dict)
- assert_true('deployment_dir' in workflow_dict, workflow_dict)
- assert_true('schema_version' in workflow_dict, workflow_dict)
- assert_true('job_properties' in workflow_dict, workflow_dict)
- assert_true('start' in workflow_dict, workflow_dict)
- assert_true('nodes' in workflow_dict, workflow_dict)
- assert_true('id' in workflow_dict, workflow_dict)
- assert_true('name' in workflow_dict, workflow_dict)
- # Check links
- for node in workflow_dict['nodes']:
- assert_true('child_links' in node, node)
- for link in node['child_links']:
- assert_true('name' in link, link)
- assert_true('comment' in link, link)
- assert_true('parent' in link, link)
- assert_true('child' in link, link)
- def test_model_to_dict(self):
- node_dict = model_to_dict(self.wf.node_set.filter(node_type='start')[0])
- # Test properties
- assert_true('id' in node_dict)
- assert_true('name' in node_dict)
- assert_true('description' in node_dict)
- assert_true('node_type' in node_dict)
- assert_true('workflow' in node_dict)
- # Utils
- WORKFLOW_DICT = {u'deployment_dir': [u''], u'name': [u'wf-name-1'], u'description': [u''],
- u'schema_version': [u'uri:oozie:workflow:0.2'],
- u'parameters': [u'[{"name":"market","value":"US"}]'],
- u'job_xml': [u'jobconf.xml'],
- u'job_properties': [u'[{"name":"sleep-all","value":"${SLEEP}"}]']
- }
- COORDINATOR_DICT = {u'name': [u'MyCoord'], u'description': [u'Description of my coodinator'],
- u'workflow': [u'1'],
- u'frequency_number': [u'1'], u'frequency_unit': [u'days'],
- u'start_0': [u'07/01/2012'], u'start_1': [u'12:00 AM'],
- u'end_0': [u'07/04/2012'], u'end_1': [u'12:00 AM'],
- u'timezone': [u'America/Los_Angeles'],
- u'parameters': [u'[{"name":"market","value":"US,France"}]'],
- u'timeout': [u'100'],
- u'concurrency': [u'3'],
- u'execution': [u'FIFO'],
- u'throttle': [u'10'],
- u'schema_version': [u'uri:oozie:coordinator:0.1']
- }
- def remove_related_fields(workflow_dict):
- """
- workflow_dict is a workflow that has been converted into a dictionary via workflow_to_dict
- """
- del workflow_dict['owner']
- del workflow_dict['job_ptr']
- for node in workflow_dict['nodes']:
- del node['node_ptr']
- return workflow_dict
- def add_node(workflow, name, node_type, parents, attrs={}):
- """
- create a node of type node_type and associate the listed parents.
- """
- NodeClass = NODE_TYPES[node_type]
- node = NodeClass(workflow=workflow, node_type=node_type, name=name)
- for attr in attrs:
- setattr(node, attr, attrs[attr])
- node.save()
- # Add parent
- # If skipped, remember to preserve order: regular links first, then error link
- if parents:
- for parent in parents:
- name = 'ok'
- if parent.node_type == 'start' or parent.node_type == 'join':
- name = 'to'
- elif parent.node_type == 'fork' or parent.node_type == 'decision':
- name = 'start'
- link = Link(parent=parent, child=node, name=name)
- link.save()
- # Create error link
- if node_type != 'fork' and node_type != 'decision' and node_type != 'join':
- link = Link(parent=node, child=Kill.objects.get(name='kill', workflow=workflow), name="error")
- link.save()
- return node
- def create_workflow(client, workflow_dict=WORKFLOW_DICT):
- name = str(workflow_dict['name'][0])
- Node.objects.filter(workflow__name=name).delete()
- Workflow.objects.filter(name=name).delete()
- workflow_count = Workflow.objects.count()
- response = client.get(reverse('oozie:create_workflow'))
- assert_equal(workflow_count, Workflow.objects.count(), response)
- response = client.post(reverse('oozie:create_workflow'), workflow_dict, follow=True)
- assert_equal(200, response.status_code)
- assert_equal(workflow_count + 1, Workflow.objects.count(), response)
- wf = Workflow.objects.get(name=name)
- assert_not_equal('', wf.deployment_dir)
- return wf
- def create_coordinator(workflow, client):
- coord_count = Coordinator.objects.count()
- response = client.get(reverse('oozie:create_coordinator'))
- assert_equal(coord_count, Coordinator.objects.count(), response)
- post = COORDINATOR_DICT.copy()
- post['workflow'] = workflow.id
- response = client.post(reverse('oozie:create_coordinator'), post)
- assert_equal(coord_count + 1, Coordinator.objects.count(), response)
- return Coordinator.objects.get(name='MyCoord')
- def create_dataset(coord, client):
- response = client.post(reverse('oozie:create_coordinator_dataset', args=[coord.id]), {
- u'create-name': [u'MyDataset'], u'create-frequency_number': [u'1'], u'create-frequency_unit': [u'days'],
- u'create-uri': [u'/data/${YEAR}${MONTH}${DAY}'],
- u'create-start_0': [u'07/01/2012'], u'create-start_1': [u'12:00 AM'],
- u'create-timezone': [u'America/Los_Angeles'], u'create-done_flag': [u''],
- u'create-description': [u'']})
- data = json.loads(response.content)
- assert_equal(0, data['status'], data['data'])
- def create_coordinator_data(coord, client):
- response = client.post(reverse('oozie:create_coordinator_data', args=[coord.id, 'input']),
- {u'input-name': [u'input_dir'], u'input-dataset': [u'1']})
- data = json.loads(response.content)
- assert_equal(0, data['status'], data['data'])
|