a11086ffac
* feat: allow editing html block imported from upstream The modified field is left untouched in future sync while storing the upstream values in hidden fields to allow authors to revert to upstream version at any point. * fix: sync downstream_customized field for copy-pasted modified block * test: add more tests * fix: lint issues * test: copy paste * feat: skip sync if html data is modified * feat: update upstream fields only when modified * refactor: use version_synced field to skip sync * feat: edit title inplace for library source components * fixup! feat: edit title inplace for library source components * fix: edit title button style * fix: test case * fix: lint issue * refactor: don't show different icon for modified upstream blocks * Revert "refactor: use version_synced field to skip sync" This reverts commit 8b784fff2f49b43702c952e7f955bd4048e8cc69. * feat: only skip sync for modified blocks if updated as part of container * refactor: update sync behaviour when synced individually and as part of parent * feat: include ready to sync children info in downstream link get api * test: fix failing tests * fix: lint issues * feat: new tests and update api to allow overriding modified fields in sync * test: api changes * refactor: edit options should be visible for individual imports * docs: update api docs * chore: remove old comments
644 lines
28 KiB
Python
644 lines
28 KiB
Python
"""
|
|
Test CMS's upstream->downstream syncing system
|
|
"""
|
|
import datetime
|
|
|
|
import ddt
|
|
from organizations.api import ensure_organization
|
|
from organizations.models import Organization
|
|
from pytz import utc
|
|
|
|
from cms.lib.xblock.upstream_sync import (
|
|
BadDownstream,
|
|
BadUpstream,
|
|
NoUpstream,
|
|
UpstreamLink,
|
|
decline_sync,
|
|
sever_upstream_link,
|
|
)
|
|
from cms.lib.xblock.upstream_sync_block import sync_from_upstream_block, fetch_customizable_fields_from_block
|
|
from cms.djangoapps.contentstore.xblock_storage_handlers.view_handlers import save_xblock_with_callback
|
|
from common.djangoapps.student.tests.factories import UserFactory
|
|
from openedx.core.djangoapps.content_libraries import api as libs
|
|
from openedx.core.djangoapps.content_tagging import api as tagging_api
|
|
from openedx.core.djangoapps.xblock import api as xblock
|
|
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
|
|
from xmodule.modulestore.tests.factories import BlockFactory, CourseFactory
|
|
|
|
|
|
@ddt.ddt
|
|
class UpstreamTestCase(ModuleStoreTestCase):
|
|
"""
|
|
Tests the upstream_sync mixin, data object, and Python APIs.
|
|
"""
|
|
|
|
def setUp(self):
|
|
"""
|
|
Create a simple course with one unit, and simple V2 library with two blocks.
|
|
"""
|
|
super().setUp()
|
|
course = CourseFactory.create()
|
|
chapter = BlockFactory.create(category='chapter', parent=course)
|
|
sequential = BlockFactory.create(category='sequential', parent=chapter)
|
|
self.unit = BlockFactory.create(category='vertical', parent=sequential)
|
|
|
|
ensure_organization("TestX")
|
|
self.library = libs.create_library(
|
|
org=Organization.objects.get(short_name="TestX"),
|
|
slug="TestLib",
|
|
title="Test Upstream Library",
|
|
)
|
|
self.upstream_key = libs.create_library_block(self.library.key, "html", "test-upstream").usage_key
|
|
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
upstream.display_name = "Upstream Title V2"
|
|
upstream.data = "<html><body>Upstream content V2</body></html>"
|
|
upstream.save()
|
|
|
|
self.upstream_problem_key = libs.create_library_block(self.library.key, "problem", "problem-upstream").usage_key
|
|
libs.set_library_block_olx(self.upstream_problem_key, (
|
|
'<problem'
|
|
' attempts_before_showanswer_button="1"'
|
|
' display_name="Upstream Problem Title V2"'
|
|
' due="2024-01-01T00:00:00Z"'
|
|
' force_save_button="false"'
|
|
' graceperiod="1d"'
|
|
' grading_method="last_attempt"'
|
|
' matlab_api_key="abc"'
|
|
' max_attempts="10"'
|
|
' rerandomize=""always""'
|
|
' show_correctness="never"'
|
|
' show_reset_button="false"'
|
|
' showanswer="on_correct"'
|
|
' submission_wait_seconds="10"'
|
|
' use_latex_compiler="false"'
|
|
' weight="1"'
|
|
'/>\n'
|
|
))
|
|
|
|
self.upstream_video_key = libs.create_library_block(self.library.key, "video", "video-upstream").usage_key
|
|
libs.set_library_block_olx(self.upstream_video_key, (
|
|
'<video'
|
|
' display_name="Video Test"'
|
|
' edx_video_id=""'
|
|
' end_time="00:00:00"'
|
|
' html5_sources="["https://www.sample-videos.com/video321/mp4/720/big_buck_bunny_720p_2mb.mp4"]"'
|
|
' start_time="00:00:00"'
|
|
' track=""'
|
|
' youtube_id_1_0=""'
|
|
'>'
|
|
' <source src="https://www.sample-videos.com/video321/mp4/720/big_buck_bunny_720p_2mb.mp4"/>'
|
|
'</video>'
|
|
))
|
|
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
self.taxonomy_all_org = tagging_api.create_taxonomy(
|
|
"test_taxonomy",
|
|
"Test Taxonomy",
|
|
export_id="ALL_ORGS",
|
|
)
|
|
tagging_api.set_taxonomy_orgs(self.taxonomy_all_org, all_orgs=True)
|
|
for tag_value in ('tag_1', 'tag_2', 'tag_3', 'tag_4', 'tag_5', 'tag_6', 'tag_7'):
|
|
tagging_api.add_tag_to_taxonomy(self.taxonomy_all_org, tag_value)
|
|
|
|
self.upstream_tags = ['tag_1', 'tag_5']
|
|
tagging_api.tag_object(str(self.upstream_key), self.taxonomy_all_org, self.upstream_tags)
|
|
|
|
def test_sync_bad_downstream(self):
|
|
"""
|
|
Syncing into an unsupported downstream (such as a another Content Library block) raises BadDownstream, but
|
|
doesn't affect the block.
|
|
"""
|
|
downstream_lib_block_key = libs.create_library_block(self.library.key, "html", "bad-downstream").usage_key
|
|
downstream_lib_block = xblock.load_block(downstream_lib_block_key, self.user)
|
|
downstream_lib_block.display_name = "Another lib block"
|
|
downstream_lib_block.data = "<html>another lib block</html>"
|
|
downstream_lib_block.upstream = str(self.upstream_key)
|
|
downstream_lib_block.save()
|
|
|
|
with self.assertRaises(BadDownstream):
|
|
sync_from_upstream_block(downstream_lib_block, self.user)
|
|
|
|
assert downstream_lib_block.display_name == "Another lib block"
|
|
assert downstream_lib_block.data == "<html>another lib block</html>"
|
|
|
|
def test_sync_no_upstream(self):
|
|
"""
|
|
Trivial case: Syncing a block with no upstream is a no-op
|
|
"""
|
|
block = BlockFactory.create(category='html', parent=self.unit)
|
|
block.display_name = "Block Title"
|
|
block.data = "Block content"
|
|
|
|
with self.assertRaises(NoUpstream):
|
|
sync_from_upstream_block(block, self.user)
|
|
|
|
assert block.display_name == "Block Title"
|
|
assert block.data == "Block content"
|
|
assert not block.upstream_display_name
|
|
|
|
@ddt.data(
|
|
("not-a-key-at-all", ".*is malformed.*"),
|
|
("course-v1:Oops+ItsA+CourseKey", ".*is malformed.*"),
|
|
("block-v1:The+Wrong+KindOfUsageKey+type@html+block@nope", ".*is malformed.*"),
|
|
("lb:TestX:NoSuchLib:html:block-id", ".*not found in the system.*"),
|
|
("lb:TestX:TestLib:html:no-such-html", ".*not found in the system.*"),
|
|
)
|
|
@ddt.unpack
|
|
def test_sync_bad_upstream(self, upstream, message_regex):
|
|
"""
|
|
Syncing with a bad upstream raises BadUpstream, but doesn't affect the block
|
|
"""
|
|
block = BlockFactory.create(category='html', parent=self.unit, upstream=upstream)
|
|
block.display_name = "Block Title"
|
|
block.data = "Block content"
|
|
|
|
with self.assertRaisesRegex(BadUpstream, message_regex):
|
|
sync_from_upstream_block(block, self.user)
|
|
|
|
assert block.display_name == "Block Title"
|
|
assert block.data == "Block content"
|
|
assert not block.upstream_display_name
|
|
|
|
def test_sync_incompatible_upstream(self):
|
|
"""
|
|
Syncing with a bad upstream raises BadUpstream, but doesn't affect the block
|
|
"""
|
|
downstream_block = BlockFactory.create(
|
|
category='html', parent=self.unit, upstream=str(self.upstream_problem_key),
|
|
)
|
|
downstream_block.display_name = "Block Title"
|
|
downstream_block.data = "Block content"
|
|
|
|
with self.assertRaisesRegex(BadUpstream, "Content type mismatch.*"):
|
|
sync_from_upstream_block(downstream_block, self.user)
|
|
|
|
assert downstream_block.display_name == "Block Title"
|
|
assert downstream_block.data == "Block content"
|
|
assert not downstream_block.upstream_display_name
|
|
|
|
def test_sync_not_accessible(self):
|
|
"""
|
|
Syncing with an block that exists, but is inaccessible, raises BadUpstream
|
|
"""
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
user_who_cannot_read_upstream = UserFactory.create(username="rando", is_staff=False, is_superuser=False)
|
|
with self.assertRaisesRegex(BadUpstream, ".*could not be loaded.*") as exc:
|
|
sync_from_upstream_block(downstream, user_who_cannot_read_upstream)
|
|
|
|
def test_sync_updates_happy_path(self):
|
|
"""
|
|
Can we sync updates from a content library block to a linked out-of-date course block?
|
|
"""
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
|
|
# Initial sync
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.upstream_version == 2 # Library blocks start at version 2 (v1 is the empty new block)
|
|
assert downstream.upstream_display_name == "Upstream Title V2"
|
|
assert downstream.display_name == "Upstream Title V2"
|
|
assert downstream.data == "<html><body>Upstream content V2</body></html>"
|
|
|
|
# Verify tags
|
|
object_tags = tagging_api.get_object_tags(str(downstream.location))
|
|
assert len(object_tags) == len(self.upstream_tags)
|
|
for object_tag in object_tags:
|
|
assert object_tag.value in self.upstream_tags
|
|
|
|
# Upstream updates
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
upstream.display_name = "Upstream Title V3"
|
|
upstream.data = "<html><body>Upstream content V3</body></html>"
|
|
upstream.save()
|
|
new_upstream_tags = self.upstream_tags + ['tag_2', 'tag_3']
|
|
tagging_api.tag_object(str(self.upstream_key), self.taxonomy_all_org, new_upstream_tags)
|
|
|
|
# Assert that un-published updates are not yet pulled into downstream
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.upstream_version == 2 # Library blocks start at version 2 (v1 is the empty new block)
|
|
assert downstream.upstream_display_name == "Upstream Title V2"
|
|
assert downstream.display_name == "Upstream Title V2"
|
|
assert downstream.data == "<html><body>Upstream content V2</body></html>"
|
|
|
|
# Publish changes
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
# Follow-up sync. Assert that updates are pulled into downstream.
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.upstream_version == 3
|
|
assert downstream.upstream_display_name == "Upstream Title V3"
|
|
assert downstream.display_name == "Upstream Title V3"
|
|
assert downstream.data == "<html><body>Upstream content V3</body></html>"
|
|
|
|
# Verify tags
|
|
object_tags = tagging_api.get_object_tags(str(downstream.location))
|
|
assert len(object_tags) == len(new_upstream_tags)
|
|
for object_tag in object_tags:
|
|
assert object_tag.value in new_upstream_tags
|
|
|
|
# pylint: disable=too-many-statements
|
|
def test_sync_updates_to_downstream_only_fields(self):
|
|
"""
|
|
If we sync to modified content, will it preserve downstream-only fields, and overwrite the rest?
|
|
"""
|
|
downstream = BlockFactory.create(category='problem', parent=self.unit, upstream=str(self.upstream_problem_key))
|
|
|
|
# Initial sync
|
|
sync_from_upstream_block(downstream, self.user)
|
|
|
|
# These fields are copied from upstream
|
|
assert downstream.upstream_display_name == "Upstream Problem Title V2"
|
|
assert downstream.display_name == "Upstream Problem Title V2"
|
|
assert downstream.rerandomize == '"always"'
|
|
assert downstream.matlab_api_key == 'abc'
|
|
assert not downstream.use_latex_compiler
|
|
|
|
# These fields are "downstream only", so field defaults are preserved, and values are NOT copied from upstream
|
|
assert downstream.attempts_before_showanswer_button == 0
|
|
assert downstream.due is None
|
|
assert not downstream.force_save_button
|
|
assert downstream.graceperiod is None
|
|
assert downstream.grading_method == 'last_score'
|
|
assert downstream.max_attempts is None
|
|
assert downstream.show_correctness == 'always'
|
|
assert not downstream.show_reset_button
|
|
assert downstream.showanswer == 'finished'
|
|
assert downstream.submission_wait_seconds == 0
|
|
assert downstream.weight is None
|
|
|
|
# Upstream updates
|
|
libs.set_library_block_olx(self.upstream_problem_key, (
|
|
'<problem'
|
|
' attempts_before_showanswer_button="10"'
|
|
' display_name="Upstream Problem Title V3"'
|
|
' due="2024-02-02T00:00:00Z"'
|
|
' force_save_button="false"'
|
|
' graceperiod=""'
|
|
' grading_method="final_attempt"'
|
|
' matlab_api_key="def"'
|
|
' max_attempts="11"'
|
|
' rerandomize=""per_student""'
|
|
' show_correctness="past_due"'
|
|
' show_reset_button="false"'
|
|
' showanswer="attempted"'
|
|
' submission_wait_seconds="11"'
|
|
' use_latex_compiler="true"'
|
|
' weight="2"'
|
|
'/>\n'
|
|
))
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
# Modifing downstream-only fields are "safe" customizations
|
|
downstream.display_name = "Downstream Title Override"
|
|
downstream.attempts_before_showanswer_button = 2
|
|
downstream.due = datetime.datetime(2025, 2, 2, tzinfo=utc)
|
|
downstream.force_save_button = True
|
|
downstream.graceperiod = '2d'
|
|
downstream.grading_method = 'last_score'
|
|
downstream.max_attempts = 100
|
|
downstream.show_correctness = 'always'
|
|
downstream.show_reset_button = True
|
|
downstream.showanswer = 'on_expired'
|
|
downstream.submission_wait_seconds = 100
|
|
downstream.weight = 3
|
|
|
|
# Modifying synchronized fields are "unsafe" customizations
|
|
downstream.rerandomize = '"onreset"'
|
|
downstream.matlab_api_key = 'hij'
|
|
save_xblock_with_callback(downstream, self.user)
|
|
|
|
# Follow-up sync.
|
|
sync_from_upstream_block(downstream, self.user)
|
|
|
|
# "unsafe" customizations are overridden by upstream
|
|
assert downstream.upstream_display_name == "Upstream Problem Title V3"
|
|
assert downstream.rerandomize == '"per_student"'
|
|
assert downstream.matlab_api_key == 'def'
|
|
assert downstream.use_latex_compiler
|
|
|
|
# but "safe" customizations survive
|
|
assert downstream.display_name == "Downstream Title Override"
|
|
assert downstream.attempts_before_showanswer_button == 2
|
|
assert downstream.due == datetime.datetime(2025, 2, 2, tzinfo=utc)
|
|
assert downstream.force_save_button
|
|
assert downstream.graceperiod == '2d'
|
|
assert downstream.grading_method == 'last_score'
|
|
assert downstream.max_attempts == 100
|
|
assert downstream.show_correctness == 'always'
|
|
assert downstream.show_reset_button
|
|
assert downstream.showanswer == 'on_expired'
|
|
assert downstream.submission_wait_seconds == 100
|
|
assert downstream.weight == 3
|
|
|
|
def test_sync_updates_to_modified_content(self):
|
|
"""
|
|
If we sync to modified content, will it preserve customizable fields, but overwrite the rest?
|
|
"""
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
|
|
# Initial sync
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.upstream_display_name == "Upstream Title V2"
|
|
assert downstream.display_name == "Upstream Title V2"
|
|
assert downstream.data == "<html><body>Upstream content V2</body></html>"
|
|
|
|
# Upstream updates
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
upstream.display_name = "Upstream Title V3"
|
|
upstream.data = "<html><body>Upstream content V3</body></html>"
|
|
upstream.save()
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
# Downstream modifications, currently all fields are overridden on individual sync
|
|
downstream.display_name = "Downstream Title Override"
|
|
downstream.data = "Downstream content override"
|
|
save_xblock_with_callback(downstream, self.user)
|
|
|
|
# Follow-up sync. Assert that updates are pulled into downstream, but customizations are saved.
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.display_name == "Downstream Title Override" # "safe" customization survives
|
|
assert downstream.data == "Downstream content override" # "safe" customization survives
|
|
# Verify hidden field has latest upstream value
|
|
assert downstream.upstream_data == "<html><body>Upstream content V3</body></html>"
|
|
assert downstream.upstream_display_name == "Upstream Title V3"
|
|
|
|
def test_sync_to_downstream_with_subtle_customization(self):
|
|
"""
|
|
Edge case: If our downstream customizes a field, but then the upstream is changed to match the
|
|
customization do we still remember that the downstream field is customized? That is,
|
|
if the upstream later changes again, do we retain the downstream customization (rather than
|
|
following the upstream update?)
|
|
"""
|
|
# Start with an uncustomized downstream block.
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.downstream_customized == []
|
|
assert downstream.display_name == downstream.upstream_display_name == "Upstream Title V2"
|
|
|
|
# Then, customize our downstream title.
|
|
downstream.display_name = "Title V3"
|
|
save_xblock_with_callback(downstream, self.user)
|
|
assert downstream.downstream_customized == ["display_name"]
|
|
|
|
# Syncing should retain the customization if we allow display name customization.
|
|
sync_from_upstream_block(
|
|
downstream,
|
|
self.user,
|
|
override_customizations=True,
|
|
keep_custom_fields=["display_name"]
|
|
)
|
|
assert downstream.upstream_version == 2
|
|
assert downstream.upstream_display_name == "Upstream Title V2"
|
|
assert downstream.display_name == "Title V3"
|
|
|
|
# Whoa, look at that, the upstream has updated itself to the exact same title...
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
upstream.display_name = "Title V3"
|
|
upstream.save()
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
# ...which is reflected when we sync.
|
|
sync_from_upstream_block(
|
|
downstream,
|
|
self.user,
|
|
override_customizations=True,
|
|
keep_custom_fields=["display_name"]
|
|
)
|
|
assert downstream.upstream_version == 3
|
|
assert downstream.upstream_display_name == downstream.display_name == "Title V3"
|
|
|
|
# But! Our downstream knows that its title is still customized.
|
|
assert downstream.downstream_customized == ["display_name"]
|
|
# So, if the upstream title changes again...
|
|
upstream.display_name = "Title V4"
|
|
upstream.save()
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
# ...then the downstream title should remain put.
|
|
sync_from_upstream_block(
|
|
downstream,
|
|
self.user,
|
|
override_customizations=True,
|
|
keep_custom_fields=["display_name"]
|
|
)
|
|
assert downstream.upstream_version == 4
|
|
assert downstream.upstream_display_name == "Title V4"
|
|
assert downstream.display_name == "Title V3"
|
|
|
|
# Finally, if we don't allow keeping any customizations
|
|
upstream.display_name = "Title V5"
|
|
upstream.save()
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
sync_from_upstream_block(
|
|
downstream,
|
|
self.user,
|
|
override_customizations=True,
|
|
keep_custom_fields=[]
|
|
) # No customizations!
|
|
assert downstream.upstream_version == 5
|
|
assert downstream.upstream_display_name == downstream.display_name == "Title V5"
|
|
# Clears downstream_customized field as well
|
|
assert downstream.downstream_customized == []
|
|
|
|
@ddt.data(None, "Title From Some Other Upstream Version")
|
|
def test_update_customizable_fields(self, initial_upstream_display_name):
|
|
"""
|
|
Can we fetch a block's upstream field values without syncing it?
|
|
|
|
Test both with and without a pre-"fetched" upstrema values on the downstream.
|
|
"""
|
|
downstream = BlockFactory.create(category='html', parent=self.unit)
|
|
downstream.upstream_display_name = initial_upstream_display_name
|
|
downstream.display_name = "Some Title"
|
|
downstream.data = "<html><data>Some content</data></html>"
|
|
|
|
# Note that we're not linked to any upstream. fetch_customizable_fields_from_block shouldn't care.
|
|
assert not downstream.upstream
|
|
assert not downstream.upstream_version
|
|
|
|
# fetch!
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
fetch_customizable_fields_from_block(upstream=upstream, downstream=downstream, user=self.user)
|
|
|
|
# Ensure: fetching doesn't affect the upstream link (or lack thereof).
|
|
assert not downstream.upstream
|
|
assert not downstream.upstream_version
|
|
|
|
# Ensure: fetching doesn't affect actual content or settings.
|
|
assert downstream.display_name == "Some Title"
|
|
assert downstream.data == "<html><data>Some content</data></html>"
|
|
|
|
# Ensure: fetching DOES set the upstream_* fields.
|
|
assert downstream.upstream_display_name == "Upstream Title V2"
|
|
|
|
def test_prompt_and_decline_sync(self):
|
|
"""
|
|
Is the user prompted for sync when it's available? Does declining remove the prompt until a new sync is ready?
|
|
"""
|
|
# Initial conditions (pre-sync)
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
link = UpstreamLink.get_for_block(downstream)
|
|
assert link.version_synced is None
|
|
assert link.version_declined is None
|
|
assert link.version_available == 2 # Library block with content starts at version 2
|
|
assert link.ready_to_sync is True
|
|
|
|
# Initial sync to V2
|
|
sync_from_upstream_block(downstream, self.user)
|
|
link = UpstreamLink.get_for_block(downstream)
|
|
assert link.version_synced == 2
|
|
assert link.version_declined is None
|
|
assert link.version_available == 2
|
|
assert link.ready_to_sync is False
|
|
|
|
# Upstream updated to V3, but not yet published
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
upstream.data = "<html><body>Upstream content V3</body></html>"
|
|
upstream.save()
|
|
link = UpstreamLink.get_for_block(downstream)
|
|
assert link.version_synced == 2
|
|
assert link.version_declined is None
|
|
assert link.version_available == 2
|
|
assert link.ready_to_sync is False
|
|
|
|
# Publish changes
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
link = UpstreamLink.get_for_block(downstream)
|
|
assert link.version_synced == 2
|
|
assert link.version_declined is None
|
|
assert link.version_available == 3
|
|
assert link.ready_to_sync is True
|
|
|
|
# Decline to sync to V3 -- ready_to_sync becomes False.
|
|
decline_sync(downstream)
|
|
link = UpstreamLink.get_for_block(downstream)
|
|
assert link.version_synced == 2
|
|
assert link.version_declined == 3
|
|
assert link.version_available == 3
|
|
assert link.ready_to_sync is False
|
|
|
|
# Upstream updated to V4 -- ready_to_sync becomes True again.
|
|
upstream = xblock.load_block(self.upstream_key, self.user)
|
|
upstream.data = "<html><body>Upstream content V4</body></html>"
|
|
upstream.save()
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
link = UpstreamLink.get_for_block(downstream)
|
|
assert link.version_synced == 2
|
|
assert link.version_declined == 3
|
|
assert link.version_available == 4
|
|
assert link.ready_to_sync is True
|
|
|
|
def test_sever_upstream_link(self):
|
|
"""
|
|
Does sever_upstream_link correctly disconnect a block from its upstream?
|
|
"""
|
|
# Start with a course block that is linked+synced to a content library block.
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
sync_from_upstream_block(downstream, self.user)
|
|
|
|
# (sanity checks)
|
|
assert downstream.upstream == str(self.upstream_key)
|
|
assert downstream.upstream_version == 2
|
|
assert downstream.upstream_display_name == "Upstream Title V2"
|
|
assert downstream.display_name == "Upstream Title V2"
|
|
assert downstream.data == "<html><body>Upstream content V2</body></html>"
|
|
assert downstream.copied_from_block is None
|
|
|
|
# Now, disconnect the course block.
|
|
sever_upstream_link(downstream)
|
|
|
|
# All upstream metadata has been wiped out.
|
|
assert downstream.upstream is None
|
|
assert downstream.upstream_version is None
|
|
assert downstream.upstream_display_name is None
|
|
|
|
# BUT, the content which was synced into the upstream remains.
|
|
assert downstream.display_name == "Upstream Title V2"
|
|
assert downstream.data == "<html><body>Upstream content V2</body></html>"
|
|
|
|
# AND, we have recorded the old upstream as our copied_from_block.
|
|
assert downstream.copied_from_block == str(self.upstream_key)
|
|
|
|
def test_sync_library_block_tags(self):
|
|
upstream_lib_block_key = libs.create_library_block(self.library.key, "html", "upstream").usage_key
|
|
upstream_lib_block = xblock.load_block(upstream_lib_block_key, self.user)
|
|
upstream_lib_block.display_name = "Another lib block"
|
|
upstream_lib_block.data = "<html>another lib block</html>"
|
|
upstream_lib_block.save()
|
|
|
|
libs.publish_changes(self.library.key, self.user.id)
|
|
|
|
expected_tags = self.upstream_tags
|
|
tagging_api.tag_object(str(upstream_lib_block_key), self.taxonomy_all_org, expected_tags)
|
|
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(upstream_lib_block_key))
|
|
|
|
# Initial sync
|
|
sync_from_upstream_block(downstream, self.user)
|
|
|
|
# Verify tags
|
|
object_tags = tagging_api.get_object_tags(str(downstream.location))
|
|
assert len(object_tags) == len(expected_tags)
|
|
for object_tag in object_tags:
|
|
assert object_tag.value in expected_tags
|
|
|
|
# Upstream updates
|
|
upstream_lib_block.display_name = "Upstream Title V3"
|
|
upstream_lib_block.data = "<html><body>Upstream content V3</body></html>"
|
|
upstream_lib_block.save()
|
|
new_upstream_tags = self.upstream_tags + ['tag_2', 'tag_3']
|
|
tagging_api.tag_object(str(upstream_lib_block_key), self.taxonomy_all_org, new_upstream_tags)
|
|
|
|
# Follow-up sync.
|
|
sync_from_upstream_block(downstream, self.user)
|
|
|
|
#Verify tags
|
|
object_tags = tagging_api.get_object_tags(str(downstream.location))
|
|
assert len(object_tags) == len(new_upstream_tags)
|
|
for object_tag in object_tags:
|
|
assert object_tag.value in new_upstream_tags
|
|
|
|
def test_sync_video_block(self):
|
|
downstream = BlockFactory.create(category='video', parent=self.unit, upstream=str(self.upstream_video_key))
|
|
downstream.edx_video_id = "test_video_id"
|
|
|
|
# Sync
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.upstream_version == 2
|
|
assert downstream.upstream_display_name == "Video Test"
|
|
assert downstream.display_name == "Video Test"
|
|
|
|
# `edx_video_id` doesn't change
|
|
assert downstream.edx_video_id == "test_video_id"
|
|
|
|
def test_sync_keep_customizaton_option(self):
|
|
"""
|
|
Test that when an upstream block has a customized downstream block, we keep
|
|
the customized options when syncing based on keep_custom_fields option.
|
|
"""
|
|
# Start with an uncustomized downstream block.
|
|
downstream = BlockFactory.create(category='html', parent=self.unit, upstream=str(self.upstream_key))
|
|
sync_from_upstream_block(downstream, self.user)
|
|
assert downstream.downstream_customized == []
|
|
assert downstream.display_name == downstream.upstream_display_name == "Upstream Title V2"
|
|
|
|
# Then, customize our downstream title and content
|
|
downstream.display_name = "Title V3"
|
|
downstream.data = "<html><data>Some content</data></html>"
|
|
save_xblock_with_callback(downstream, self.user)
|
|
assert downstream.downstream_customized == ["display_name", "data"]
|
|
|
|
# Now, sync the upstream block with `keep_custom_fields=["display_name"] only`.
|
|
# And let data be overridden
|
|
sync_from_upstream_block(
|
|
downstream,
|
|
self.user,
|
|
override_customizations=True,
|
|
keep_custom_fields=['display_name']
|
|
)
|
|
assert downstream.display_name == "Title V3"
|
|
# data is overridden
|
|
assert downstream.data == "<html><body>Upstream content V2</body></html>"
|
|
assert downstream.downstream_customized == ["display_name"]
|