diff --git a/common/lib/xmodule/xmodule/modulestore/LocMapperStore.py b/common/lib/xmodule/xmodule/modulestore/LocMapperStore.py new file mode 100644 index 0000000000..b71f647a27 --- /dev/null +++ b/common/lib/xmodule/xmodule/modulestore/LocMapperStore.py @@ -0,0 +1,397 @@ +''' +Method for converting among our differing Location/Locator whatever reprs +''' +from __future__ import absolute_import +from random import randint +import re +import pymongo +from django.conf import settings + +from xmodule.modulestore.exceptions import InvalidLocationError, ItemNotFoundError, DuplicateItemError +from xmodule.modulestore.locator import BlockUsageLocator +from xmodule.modulestore.mongo import draft +from xmodule.modulestore import Location + +def loc_mapper(): + """ + Get the loc mapper which bidirectionally maps Locations to Locators. Used like modulestore() as + a singleton accessor. + """ + # pylint: disable=W0212 + if LocMapperStore._singleton is None: + # instantiate + LocMapperStore(settings.modulestore_options) + return LocMapperStore._singleton + +class LocMapperStore(object): + ''' + This store persists mappings among the addressing schemes. At this time, it's between the old i4x Location + tuples and the split mongo Course and Block Locator schemes. + + edX has used several different addressing schemes. The original ones were organically created based on + immediate needs and were overly restrictive esp wrt course ids. These were slightly extended to support + some types of blocks may need to have draft states during editing to keep live courses from seeing the wip. + A later refactoring generalized course ids to enable governance and more complex naming, branch naming with + anything able to be in any branch. + + The expectation is that the configuration will have this use the same store as whatever is the default + or dominant store, but that's not a requirement. This store creates its own connection. + ''' + + _singleton = None + + def __init__(self, host, db, collection, port=27017, user=None, password=None, **kwargs): + ''' + Constructor + ''' + self.db = pymongo.database.Database(pymongo.MongoClient( + host=host, + port=port, + tz_aware=True, + **kwargs + ), db) + if user is not None and password is not None: + self.db.authenticate(user, password) + + self.location_map = self.db[collection + '.location_map'] + self.location_map.write_concern = {'w': 1} + LocMapperStore._singleton = self + + + # location_map functions + def create_map_entry(self, course_location, course_id=None, draft_branch='draft', prod_branch='published', + block_map=None): + """ + Add a new entry to map this course_location to the new style CourseLocator.course_id. If course_id is not + provided, it creates the default map of using org.course.name from the location (just like course_id) if + the location.cateogry = 'course'; otherwise, it uses org.course. + + You can create more than one mapping to the + same course_id target. In that case, the reverse translate will be arbitrary (no guarantee of which wins). + The use + case for more than one mapping is to map both org/course/run and org/course to the same new course_id thus + making a default for org/course. When querying for just org/course, the translator will prefer any entry + which does not have a name in the _id; otherwise, it will return an arbitrary match. + + Note: the opposite is not true. That is, it never makes sense to use 2 different CourseLocator.course_id + keys to index the same old Locator org/course/.. pattern. There's no checking to ensure you don't do this. + + NOTE: if there's already an entry w the given course_location, this may either overwrite that entry or + throw an error depending on how mongo is configured. + + :param course_location: a Location preferably whose category is 'course'. Unlike the other + map methods, this one doesn't take the old-style course_id because it's assumed to be called with + a course location not a block location. + :param course_id: the CourseLocator style course_id + :param draft_branch: the branch name to assign for drafts. This is hardcoded because old mongo had + a fixed notion that there was 2 and only 2 versions for modules: draft and production. The old mongo + did not, however, require that a draft version exist. The new one, however, does require a draft to + exist. + :param prod_branch: the branch name to assign for the production (live) copy. In old mongo, every course + had to have a production version (whereas new split mongo does not require that until the author's ready + to publish). + :param block_map: an optional map to specify preferred names for blocks where the keys are the + Location block names and the values are the BlockUsageLocator.block_id. + """ + if course_id is None: + if course_location.category == 'course': + course_id = "{0.org}.{0.course}.{0.name}".format(course_location) + else: + course_id = "{0.org}.{0.course}".format(course_location) + # very like _interpret_location_id but w/o the _id + location_id = {'org': course_location.org, 'course': course_location.course} + if course_location.category == 'course': + location_id['name'] = course_location.name + + self.location_map.insert({ + '_id': location_id, + 'course_id': course_id, + 'draft_branch': draft_branch, + 'prod_branch': prod_branch, + 'block_map': block_map or {}, + }) + + + def translate_location(self, old_style_course_id, location, published=True, add_entry_if_missing=True): + """ + Translate the given module location to a Locator. If the mapping has the run id in it, then you + should provide old_style_course_id with that run id in it to disambiguate the mapping if there exists more + than one entry in the mapping table for the org.course. + + The rationale for auto adding entries was that there should be a reasonable default translation + if the code just trips into this w/o creating translations. The downfall is that ambiguous course + locations may generate conflicting block_ids. + + :param old_style_course_id: the course_id used in old mongo not the new one (optional, will use location) + :param location: a Location pointing to a module + :param published: a boolean to indicate whether the caller wants the draft or published branch. + :param add_entry_if_missing: a boolean as to whether to return None or to create an entry if the course + or block is not found in the map. + + NOTE: unlike old mongo, draft branches contain the whole course; so, it applies to all category + of locations including course. + """ + location_id = self._interpret_location_course_id(old_style_course_id, location) + + maps = self.location_map.find(location_id) + if maps.count() == 0: + if add_entry_if_missing: + # create a new map + course_location = location.replace(category='course', name=location_id['_id.name']) + self.create_map_entry(course_location) + entry = self.location_map.find_one(location_id) + else: + return None + elif maps.count() > 1: + # if more than one, prefer the one w/o a name if that exists. Otherwise, choose the first (arbitrary) + # a bit odd b/c maps is a cursor and doesn't allow multitraversal w/o requerying db + entry = None + for candidate in maps: + if entry is None: + entry = candidate # pick off first ele in case we don't find one w/o a name + if 'name' not in candidate['_id']: + entry = candidate + break + else: + entry = maps[0] + + if published: + branch = entry['prod_branch'] + else: + branch = entry['draft_branch'] + + usage_id = entry['block_map'].get(location.name) + if usage_id is None: + if add_entry_if_missing: + usage_id = self._add_to_block_map(location, location_id, entry['block_map']) + else: + return None + elif isinstance(usage_id, dict): + # name is not unique, look through for the right category + if location.category in usage_id: + usage_id = usage_id[location.category] + elif add_entry_if_missing: + usage_id = self._add_to_block_map(location, location_id, entry['block_map']) + else: + return None + else: + raise InvalidLocationError() + + return BlockUsageLocator(course_id=entry['course_id'], branch=branch, usage_id=usage_id) + + def translate_locator_to_location(self, locator): + """ + Returns an old style Location for the given Locator if there's an appropriate entry in the + mapping collection. Note, it requires that the course was previously mapped (a side effect of + translate_location or explicitly via create_map_entry) and + the block's usage_id was previously stored in the + map (a side effect of translate_location or via add|update_block_location). + + If there are no matches, it returns None. + + If there's more than one location to locator mapping to the same course_id, it looks for the first + one with a mapping for the block usage_id and picks that arbitrary course location. + + :param locator: a BlockUsageLocator + """ + # Does not use _lookup_course b/c it doesn't actually require that the course exist in the active_version + # only that it has a mapping entry. + maps = self.location_map.find({'course_id': locator.course_id}) + # look for one which maps to this block usage_id + if maps.count() == 0: + return None + for candidate in maps: + for old_name, cat_to_usage in candidate['block_map'].iteritems(): + for category, usage_id in cat_to_usage.iteritems(): + if usage_id == locator.usage_id: + # figure out revision + # enforce the draft only if category in [..] logic + if category in draft.DIRECT_ONLY_CATEGORIES: + revision = None + elif locator.branch == candidate['draft_branch']: + revision = draft.DRAFT + else: + revision = None + return Location( + 'i4x', + candidate['_id']['org'], + candidate['_id']['course'], + category, + old_name, + revision) + return None + + + def add_block_location_translator(self, location, old_course_id=None, usage_id=None): + """ + Similar to translate_location which adds an entry if none is found, but this cannot create a new + course mapping entry, only a block within such a mapping entry. If it finds no existing + course maps, it raises ItemNotFoundError. + + In the case that there are more than one mapping record for the course identified by location, this + method adds the mapping to all matching records! (translate_location only adds to one) + + It allows the caller to specify + the new-style usage_id for the target rather than having the translate concoct its own. + If the provided usage_id already exists in one of the found maps for the org/course, this function + raises DuplicateItemError unless the old item id == the new one. + + If the caller does not provide a usage_id and there exists an entry in one of the course variants, + it will use that entry. If more than one variant uses conflicting entries, it will raise DuplicateItemError. + + Returns the usage_id used in the mapping + + :param location: a fully specified Location + :param old_course_id: the old-style org/course or org/course/run string (optional) + :param usage_id: the desired new block_id. If left as None, this will generate one as per translate_location + """ + location_id = self._interpret_location_course_id(old_course_id, location) + + maps = self.location_map.find(location_id) + if maps.count() == 0: + raise ItemNotFoundError() + + # turn maps from cursor to list + map_list = [map_entry for map_entry in maps] + # check whether there's already a usage_id for this location (and it agrees w/ any passed in or found) + for map_entry in map_list: + if (location.name in map_entry['block_map'] and + location.category in map_entry['block_map'][location.name]): + if usage_id is None: + usage_id = map_entry['block_map'][location.name][location.category] + elif usage_id != map_entry['block_map'][location.name][location.category]: + raise DuplicateItemError() + + computed_usage_id = usage_id + + # update the maps (and generate a usage_id if it's not been set yet) + for map_entry in map_list: + if computed_usage_id is None: + computed_usage_id = self._add_to_block_map(location, location_id, map_entry['block_map']) + elif (location.name not in map_entry['block_map'] or + location.category not in map_entry['block_map'][location.name]): + alt_usage_id = self._verify_uniqueness(computed_usage_id, map_entry['block_map']) + if alt_usage_id != computed_usage_id: + if usage_id is not None: + raise DuplicateItemError() + else: + # revise already set ones and add to remaining ones + computed_usage_id = self.update_block_location_translator( + location, + alt_usage_id, + old_course_id, + True + ) + + map_entry['block_map'].setdefault(location.name, {})[location.category] = computed_usage_id + self.location_map.update({'_id': map_entry['_id']}, {'$set': {'block_map': map_entry['block_map']}}) + + return computed_usage_id + + + def update_block_location_translator(self, location, usage_id, old_course_id=None, autogenerated_usage_id=False): + """ + Update all existing maps from location's block to the new usage_id. Used for changing the usage_id, + thus the usage_id is required. + + Returns the usage_id. (which is primarily useful in the case of autogenerated_usage_id) + + :param location: a fully specified Location + :param usage_id: the desired new block_id. + :param old_course_id: the old-style org/course or org/course/run string (optional) + :param autogenerated_usage_id: a flag used mostly for internal calls to indicate that this usage_id + was autogenerated and thus can be overridden if it's not unique. If you set this flag, the stored + usage_id may not be the one you submitted. + """ + location_id = self._interpret_location_course_id(old_course_id, location) + + maps = self.location_map.find(location_id) + for map_entry in maps: + # handle noop of renaming to same name + if (location.name in map_entry['block_map'] and + map_entry['block_map'][location.name].get(location.category) == usage_id): + continue + alt_usage_id = self._verify_uniqueness(usage_id, map_entry['block_map']) + if alt_usage_id != usage_id: + if autogenerated_usage_id: + # revise already set ones and add to remaining ones + usage_id = self.update_block_location_translator(location, alt_usage_id, old_course_id, True) + return usage_id + else: + raise DuplicateItemError() + + if location.category in map_entry['block_map'].setdefault(location.name, {}): + map_entry['block_map'][location.name][location.category] = usage_id + self.location_map.update({'_id': map_entry['_id']}, {'$set': {'block_map': map_entry['block_map']}}) + + return usage_id + + + def delete_block_location_translator(self, location, old_course_id=None): + """ + Remove all existing maps from location's block. + + :param location: a fully specified Location + :param old_course_id: the old-style org/course or org/course/run string (optional) + """ + location_id = self._interpret_location_course_id(old_course_id, location) + + maps = self.location_map.find(location_id) + for map_entry in maps: + if location.category in map_entry['block_map'].setdefault(location.name, {}): + if len(map_entry['block_map'][location.name]) == 1: + del map_entry['block_map'][location.name] + else: + del map_entry['block_map'][location.name][location.category] + self.location_map.update({'_id': map_entry['_id']}, {'$set': {'block_map': map_entry['block_map']}}) + + def _add_to_block_map(self, location, location_id, block_map): + '''add the given location to the block_map and persist it''' + if self._block_id_is_guid(location.name): + # I'm having second thoughts about this even though it will make the ids more meaningful. + # The downside is that if there's more than one course mapped to from the same org/course root + # the block ids will likely be out of sync and collide from an id perspective. HOWEVER, + # if there are few == org/course roots or their content is unrelated, this will work well. + usage_id = self._verify_uniqueness(location.category + location.name[:3], block_map) + block_map.setdefault(location.name, {})[location.category] = usage_id + self.location_map.update(location_id, {'$set': {'block_map': block_map}}) + return usage_id + + def _interpret_location_course_id(self, course_id, location): + """ + Take the old style course id (org/course/run) and return a dict for querying the mapping table. + If the course_id is empty, it uses location, but this may result in an inadequate id. + + :param course_id: old style 'org/course/run' id from Location.course_id where Location.category = 'course' + :param location: a Location object which may be to a module or a course. Provides partial info + if course_id is omitted. + """ + if course_id: + # re doesn't allow ?P<_id.org> and ilk + m = re.match(r'([^/]+)/([^/]+)/([^/]+)', course_id) + return dict(zip(['_id.org', '_id.course', '_id.name'], m.groups())) + + location_id = {'_id.org': location.org, '_id.course': location.course} + if location.category == 'course': + location_id['_id.name'] = location.name + return location_id + + def _block_id_is_guid(self, name): + return len(name) == 32 and re.search(r'[^0-9A-Fa-f]', name) is None + + def _verify_uniqueness(self, name, block_map): + ''' + Verify that the name doesn't occur elsewhere in block_map. If it does, keep adding to it until + it's unique. + ''' + for targets in block_map.itervalues(): + if isinstance(targets, dict): + for values in targets.itervalues(): + if values == name: + name += str(randint(0, 9)) + return self._verify_uniqueness(name, block_map) + + elif targets == name: + name += str(randint(0, 9)) + return self._verify_uniqueness(name, block_map) + return name diff --git a/common/lib/xmodule/xmodule/modulestore/tests/test_location_mapper.py b/common/lib/xmodule/xmodule/modulestore/tests/test_location_mapper.py index dd55927f10..f4addd9ee9 100644 --- a/common/lib/xmodule/xmodule/modulestore/tests/test_location_mapper.py +++ b/common/lib/xmodule/xmodule/modulestore/tests/test_location_mapper.py @@ -5,10 +5,10 @@ Created on Aug 5, 2013 ''' import unittest import uuid -from xmodule.modulestore.split_mongo.split import SplitMongoModuleStore from xmodule.modulestore import Location from xmodule.modulestore.locator import BlockUsageLocator from xmodule.modulestore.exceptions import ItemNotFoundError, DuplicateItemError +from xmodule.modulestore.LocMapperStore import LocMapperStore class TestLocationMapper(unittest.TestCase): @@ -18,29 +18,23 @@ class TestLocationMapper(unittest.TestCase): 'host': 'localhost', 'db': 'test_xmodule', 'collection': 'modulestore{0}'.format(uuid.uuid4().hex), - 'fs_root': '', - 'render_template': render_to_template_mock, - 'default_class': 'xmodule.raw_module.RawDescriptor', } # pylint: disable=W0142 - TestLocationMapper.modulestore = SplitMongoModuleStore(**modulestore_options) + TestLocationMapper.loc_store = LocMapperStore(**modulestore_options) def tearDown(self): - db = TestLocationMapper.modulestore.db - db.drop_collection(TestLocationMapper.modulestore.course_index) - db.drop_collection(TestLocationMapper.modulestore.structures) - db.drop_collection(TestLocationMapper.modulestore.definitions) - db.drop_collection(TestLocationMapper.modulestore.location_map) + db = TestLocationMapper.loc_store.db + db.drop_collection(TestLocationMapper.loc_store.location_map) db.connection.close() - TestLocationMapper.modulestore = None + TestLocationMapper.loc_store = None def test_create_map(self): org = 'foo_org' course = 'bar_course' - modulestore().create_map_entry(Location('i4x', org, course, 'course', 'baz_run')) - entry = modulestore().location_map.find_one({ + loc_mapper().create_map_entry(Location('i4x', org, course, 'course', 'baz_run')) + entry = loc_mapper().location_map.find_one({ '_id': {'org': org, 'course': course, 'name': 'baz_run'} }) self.assertIsNotNone(entry, "Didn't find entry") @@ -49,8 +43,8 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(entry['prod_branch'], 'published') self.assertEqual(entry['block_map'], {}) - modulestore().create_map_entry(Location('i4x', org, course, 'vertical', 'baz_vert')) - entry = modulestore().location_map.find_one({ + loc_mapper().create_map_entry(Location('i4x', org, course, 'vertical', 'baz_vert')) + entry = loc_mapper().location_map.find_one({ '_id': {'org': org, 'course': course} }) self.assertIsNotNone(entry, "Didn't find entry") @@ -59,13 +53,13 @@ class TestLocationMapper(unittest.TestCase): course = 'quux_course' # oldname: {category: newname} block_map = {'abc123': {'problem': 'problem2'}} - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'problem', 'abc123', 'draft'), 'foo_org.geek_dept.quux_course.baz_run', 'wip', 'live', block_map) - entry = modulestore().location_map.find_one({ + entry = loc_mapper().location_map.find_one({ '_id': {'org': org, 'course': course} }) self.assertIsNotNone(entry, "Didn't find entry") @@ -82,7 +76,7 @@ class TestLocationMapper(unittest.TestCase): org = 'foo_org' course = 'bar_course' old_style_course_id = '{}/{}/{}'.format(org, course, 'baz_run') - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( old_style_course_id, Location('i4x', org, course, 'problem', 'abc123'), add_entry_if_missing=False @@ -91,13 +85,13 @@ class TestLocationMapper(unittest.TestCase): new_style_course_id = '{}.geek_dept.{}.baz_run'.format(org, course) block_map = {'abc123': {'problem': 'problem2'}} - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'baz_run'), new_style_course_id, block_map=block_map ) # only one course matches - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( old_style_course_id, Location('i4x', org, course, 'problem', 'abc123'), add_entry_if_missing=False @@ -106,14 +100,14 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(prob_locator.branch, 'published') self.assertEqual(prob_locator.usage_id, 'problem2') # look for w/ only the Location (works b/c there's only one possible course match) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, Location('i4x', org, course, 'problem', 'abc123'), add_entry_if_missing=False ) self.assertEqual(prob_locator.course_id, new_style_course_id) # look for non-existent problem - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, Location('i4x', org, course, 'problem', '1def23'), add_entry_if_missing=False @@ -122,12 +116,12 @@ class TestLocationMapper(unittest.TestCase): # add a distractor course block_map = {'abc123': {'problem': 'problem3'}} - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'delta_run'), '{}.geek_dept.{}.{}'.format(org, course, 'delta_run'), block_map=block_map ) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( old_style_course_id, Location('i4x', org, course, 'problem', 'abc123'), add_entry_if_missing=False @@ -135,7 +129,7 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(prob_locator.course_id, new_style_course_id) self.assertEqual(prob_locator.usage_id, 'problem2') # look for w/ only the Location (not unique; so, just verify it returns something) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, Location('i4x', org, course, 'problem', 'abc123'), add_entry_if_missing=False @@ -143,13 +137,13 @@ class TestLocationMapper(unittest.TestCase): self.assertIsNotNone(prob_locator, "couldn't find ambiguous location") # add a default course pointing to the delta_run - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'problem', '789abc123efg456'), '{}.geek_dept.{}.{}'.format(org, course, 'delta_run'), block_map=block_map ) # now the ambiguous query should return delta - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, Location('i4x', org, course, 'problem', 'abc123'), add_entry_if_missing=False @@ -158,7 +152,7 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(prob_locator.usage_id, 'problem3') # get the draft one (I'm sorry this is getting long) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, Location('i4x', org, course, 'problem', 'abc123'), published=False, @@ -178,7 +172,7 @@ class TestLocationMapper(unittest.TestCase): old_style_course_id = '{}/{}/{}'.format(org, course, 'baz_run') problem_name = 'abc123abc123abc123abc123abc123f9' location = Location('i4x', org, course, 'problem', problem_name) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( old_style_course_id, location, add_entry_if_missing=True @@ -188,7 +182,7 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(prob_locator.branch, 'published') self.assertEqual(prob_locator.usage_id, 'problemabc') # look for w/ only the Location (works b/c there's only one possible course match) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, location, add_entry_if_missing=True @@ -196,12 +190,12 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(prob_locator.course_id, new_style_course_id) # add a distractor course - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'delta_run'), '{}.geek_dept.{}.{}'.format(org, course, 'delta_run'), block_map={problem_name: {'problem': 'problem3'}} ) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( old_style_course_id, location, add_entry_if_missing=True @@ -209,7 +203,7 @@ class TestLocationMapper(unittest.TestCase): self.assertEqual(prob_locator.course_id, new_style_course_id) self.assertEqual(prob_locator.usage_id, 'problemabc') # look for w/ only the Location (not unique; so, just verify it returns something) - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, location, add_entry_if_missing=True @@ -217,13 +211,13 @@ class TestLocationMapper(unittest.TestCase): self.assertIsNotNone(prob_locator, "couldn't find ambiguous location") # add a default course pointing to the delta_run - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'problem', '789abc123efg456'), '{}.geek_dept.{}.{}'.format(org, course, 'delta_run'), block_map={problem_name: {'problem': 'problem3'}} ) # now the ambiguous query should return delta - prob_locator = modulestore().translate_location( + prob_locator = loc_mapper().translate_location( None, location, add_entry_if_missing=False @@ -243,10 +237,10 @@ class TestLocationMapper(unittest.TestCase): course_id=new_style_course_id, usage_id='problem2' ) - prob_location = modulestore().translate_locator_to_location(prob_locator) + prob_location = loc_mapper().translate_locator_to_location(prob_locator) self.assertIsNone(prob_location, 'found entry in empty map table') - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'baz_run'), new_style_course_id, block_map={ @@ -255,33 +249,33 @@ class TestLocationMapper(unittest.TestCase): } ) # only one course matches - prob_location = modulestore().translate_locator_to_location(prob_locator) + prob_location = loc_mapper().translate_locator_to_location(prob_locator) # default branch self.assertEqual(prob_location, Location('i4x', org, course, 'problem', 'abc123', None)) # explicit branch prob_locator = BlockUsageLocator(prob_locator, branch='draft') - prob_location = modulestore().translate_locator_to_location(prob_locator) + prob_location = loc_mapper().translate_locator_to_location(prob_locator) self.assertEqual(prob_location, Location('i4x', org, course, 'problem', 'abc123', 'draft')) prob_locator = BlockUsageLocator( course_id=new_style_course_id, usage_id='problem2', branch='production' ) - prob_location = modulestore().translate_locator_to_location(prob_locator) + prob_location = loc_mapper().translate_locator_to_location(prob_locator) self.assertEqual(prob_location, Location('i4x', org, course, 'problem', 'abc123', None)) # same for chapter except chapter cannot be draft in old system chap_locator = BlockUsageLocator( course_id=new_style_course_id, usage_id='chapter48f' ) - chap_location = modulestore().translate_locator_to_location(chap_locator) + chap_location = loc_mapper().translate_locator_to_location(chap_locator) self.assertEqual(chap_location, Location('i4x', org, course, 'chapter', '48f23a10395384929234')) # explicit branch chap_locator = BlockUsageLocator(chap_locator, branch='draft') - chap_location = modulestore().translate_locator_to_location(chap_locator) + chap_location = loc_mapper().translate_locator_to_location(chap_locator) self.assertEqual(chap_location, Location('i4x', org, course, 'chapter', '48f23a10395384929234')) chap_locator = BlockUsageLocator( course_id=new_style_course_id, usage_id='chapter48f', branch='production' ) - chap_location = modulestore().translate_locator_to_location(chap_locator) + chap_location = loc_mapper().translate_locator_to_location(chap_locator) self.assertEqual(chap_location, Location('i4x', org, course, 'chapter', '48f23a10395384929234')) # look for non-existent problem @@ -290,21 +284,21 @@ class TestLocationMapper(unittest.TestCase): branch='draft', usage_id='problem3' ) - prob_location = modulestore().translate_locator_to_location(prob_locator2) + prob_location = loc_mapper().translate_locator_to_location(prob_locator2) self.assertIsNone(prob_location, 'Found non-existent problem') # add a distractor course new_style_course_id = '{}.geek_dept.{}.{}'.format(org, course, 'delta_run') - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'delta_run'), new_style_course_id, block_map={'abc123': {'problem': 'problem3'}} ) - prob_location = modulestore().translate_locator_to_location(prob_locator) + prob_location = loc_mapper().translate_locator_to_location(prob_locator) self.assertEqual(prob_location, Location('i4x', org, course, 'problem', 'abc123', None)) # add a default course pointing to the delta_run - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'problem', '789abc123efg456'), new_style_course_id, block_map={'abc123': {'problem': 'problem3'}} @@ -315,7 +309,7 @@ class TestLocationMapper(unittest.TestCase): branch='production', usage_id='problem3' ) - prob_location = modulestore().translate_locator_to_location(prob_locator) + prob_location = loc_mapper().translate_locator_to_location(prob_locator) self.assertEqual(prob_location, Location('i4x', org, course, 'problem', 'abc123')) def test_add_block(self): @@ -329,48 +323,48 @@ class TestLocationMapper(unittest.TestCase): problem_name = 'abc123abc123abc123abc123abc123f9' location = Location('i4x', org, course, 'problem', problem_name) with self.assertRaises(ItemNotFoundError): - modulestore().add_block_location_translator(location) + loc_mapper().add_block_location_translator(location) with self.assertRaises(ItemNotFoundError): - modulestore().add_block_location_translator(location, old_style_course_id) + loc_mapper().add_block_location_translator(location, old_style_course_id) # w/ one matching course new_style_course_id = '{}.{}.{}'.format(org, course, 'baz_run') - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'baz_run'), new_style_course_id, ) - new_usage_id = modulestore().add_block_location_translator(location) + new_usage_id = loc_mapper().add_block_location_translator(location) self.assertEqual(new_usage_id, 'problemabc') # look it up - translated_loc = modulestore().translate_location(old_style_course_id, location, add_entry_if_missing=False) + translated_loc = loc_mapper().translate_location(old_style_course_id, location, add_entry_if_missing=False) self.assertEqual(translated_loc.course_id, new_style_course_id) self.assertEqual(translated_loc.usage_id, new_usage_id) # w/ one distractor which has one entry already new_style_course_id = '{}.geek_dept.{}.{}'.format(org, course, 'delta_run') - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'delta_run'), new_style_course_id, block_map={'48f23a10395384929234': {'chapter': 'chapter48f'}} ) # try adding the one added before - new_usage_id2 = modulestore().add_block_location_translator(location) + new_usage_id2 = loc_mapper().add_block_location_translator(location) self.assertEqual(new_usage_id, new_usage_id2) # it should be in the distractor now - new_location = modulestore().translate_locator_to_location( + new_location = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id, usage_id=new_usage_id2) ) self.assertEqual(new_location, location) # add one close to the existing chapter (cause name collision) location = Location('i4x', org, course, 'chapter', '48f23a103953849292341234567890ab') - new_usage_id = modulestore().add_block_location_translator(location) + new_usage_id = loc_mapper().add_block_location_translator(location) self.assertRegexpMatches(new_usage_id, r'^chapter48f\d') # retrievable from both courses - new_location = modulestore().translate_locator_to_location( + new_location = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id, usage_id=new_usage_id) ) self.assertEqual(new_location, location) - new_location = modulestore().translate_locator_to_location( + new_location = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id='{}.{}.{}'.format(org, course, 'baz_run'), usage_id=new_usage_id) ) self.assertEqual(new_location, location) @@ -378,16 +372,16 @@ class TestLocationMapper(unittest.TestCase): # provoke duplicate item errors location = location.replace(name='44f23a103953849292341234567890ab') with self.assertRaises(DuplicateItemError): - modulestore().add_block_location_translator(location, usage_id=new_usage_id) - new_usage_id = modulestore().add_block_location_translator(location, old_course_id=old_style_course_id) + loc_mapper().add_block_location_translator(location, usage_id=new_usage_id) + new_usage_id = loc_mapper().add_block_location_translator(location, old_course_id=old_style_course_id) other_course_old_style = '{}/{}/{}'.format(org, course, 'delta_run') - new_usage_id2 = modulestore().add_block_location_translator( + new_usage_id2 = loc_mapper().add_block_location_translator( location, old_course_id=other_course_old_style, usage_id='{}b'.format(new_usage_id) ) with self.assertRaises(DuplicateItemError): - modulestore().add_block_location_translator(location) + loc_mapper().add_block_location_translator(location) def test_update_block(self): """ @@ -396,7 +390,7 @@ class TestLocationMapper(unittest.TestCase): org = 'foo_org' course = 'bar_course' new_style_course_id = '{}.geek_dept.{}.baz_run'.format(org, course) - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'baz_run'), new_style_course_id, block_map={ @@ -406,7 +400,7 @@ class TestLocationMapper(unittest.TestCase): } ) new_style_course_id2 = '{}.geek_dept.{}.delta_run'.format(org, course) - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'delta_run'), new_style_course_id2, block_map={ @@ -417,27 +411,27 @@ class TestLocationMapper(unittest.TestCase): ) location = Location('i4x', org, course, 'problem', '1') # change in all courses to same value - modulestore().update_block_location_translator(location, 'problem1') - trans_loc = modulestore().translate_locator_to_location( + loc_mapper().update_block_location_translator(location, 'problem1') + trans_loc = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id, usage_id='problem1') ) self.assertEqual(trans_loc, location) - trans_loc = modulestore().translate_locator_to_location( + trans_loc = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id2, usage_id='problem1') ) self.assertEqual(trans_loc, location) # try to change to overwrite used usage_id location = Location('i4x', org, course, 'chapter', '48f23a10395384929234') with self.assertRaises(DuplicateItemError): - modulestore().update_block_location_translator(location, 'chapter2') + loc_mapper().update_block_location_translator(location, 'chapter2') # just change the one course - modulestore().update_block_location_translator(location, 'chapter2', '{}/{}/{}'.format(org, course, 'baz_run')) - trans_loc = modulestore().translate_locator_to_location( + loc_mapper().update_block_location_translator(location, 'chapter2', '{}/{}/{}'.format(org, course, 'baz_run')) + trans_loc = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id, usage_id='chapter2') ) self.assertEqual(trans_loc.name, '48f23a10395384929234') # but this still points to the old - trans_loc = modulestore().translate_locator_to_location( + trans_loc = loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id2, usage_id='chapter2') ) self.assertEqual(trans_loc.name, '1') @@ -450,7 +444,7 @@ class TestLocationMapper(unittest.TestCase): org = 'foo_org' course = 'bar_course' new_style_course_id = '{}.geek_dept.{}.baz_run'.format(org, course) - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'baz_run'), new_style_course_id, block_map={ @@ -460,7 +454,7 @@ class TestLocationMapper(unittest.TestCase): } ) new_style_course_id2 = '{}.geek_dept.{}.delta_run'.format(org, course) - modulestore().create_map_entry( + loc_mapper().create_map_entry( Location('i4x', org, course, 'course', 'delta_run'), new_style_course_id2, block_map={ @@ -471,22 +465,22 @@ class TestLocationMapper(unittest.TestCase): ) location = Location('i4x', org, course, 'problem', '1') # delete from all courses - modulestore().delete_block_location_translator(location) - self.assertIsNone(modulestore().translate_locator_to_location( + loc_mapper().delete_block_location_translator(location) + self.assertIsNone(loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id, usage_id='problem1') )) - self.assertIsNone(modulestore().translate_locator_to_location( + self.assertIsNone(loc_mapper().translate_locator_to_location( BlockUsageLocator(course_id=new_style_course_id2, usage_id='problem2') )) # delete from one course location = location.replace(name='abc123') - modulestore().delete_block_location_translator(location, '{}/{}/{}'.format(org, course, 'baz_run')) - self.assertIsNone(modulestore().translate_location( + loc_mapper().delete_block_location_translator(location, '{}/{}/{}'.format(org, course, 'baz_run')) + self.assertIsNone(loc_mapper().translate_location( '{}/{}/{}'.format(org, course, 'baz_run'), location, add_entry_if_missing=False )) - locator = modulestore().translate_location( + locator = loc_mapper().translate_location( '{}/{}/{}'.format(org, course, 'delta_run'), location, add_entry_if_missing=False @@ -495,8 +489,8 @@ class TestLocationMapper(unittest.TestCase): #================================== # functions to mock existing services -def modulestore(): - return TestLocationMapper.modulestore +def loc_mapper(): + return TestLocationMapper.loc_store def render_to_template_mock(*_args): pass