Skip to content

Add fields language, provider, aggregator, role_visibility to sync op #5590

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rtibbles wants to merge 1 commit into
base: hotfixes
Choose a base branch
from
Open

Add fields language, provider, aggregator, role_visibility to sync op #5590

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
247 changes: 247 additions & 0 deletions contentcuration/contentcuration/tests/test_sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from le_utils.constants import content_kinds
from le_utils.constants import file_formats
from le_utils.constants import format_presets
from le_utils.constants import roles
from le_utils.constants.labels import accessibility_categories
from le_utils.constants.labels import learning_activities
from le_utils.constants.labels import levels
Expand All @@ -17,6 +18,7 @@
from contentcuration.models import Channel
from contentcuration.models import ContentTag
from contentcuration.models import File
from contentcuration.models import Language
from contentcuration.models import License
from contentcuration.tests import testdata
from contentcuration.tests.base import StudioAPITestCase
Expand Down Expand Up @@ -429,6 +431,251 @@ def test_sync_channel_other_metadata_labels(self):
for key, value in labels.items():
self.assertEqual(getattr(target_child, key), {value: True})

def test_sync_language_field(self):
"""
Test that the language field is synced correctly when sync_resource_details is True.
"""
self.assertFalse(self.channel.has_changes())
self.assertFalse(self.derivative_channel.has_changes())

contentnode = (
self.channel.main_tree.get_descendants()
.exclude(kind_id=content_kinds.TOPIC)
.first()
)

target_child = self.derivative_channel.main_tree.get_descendants().get(
source_node_id=contentnode.node_id
)

self.assertIsNotNone(target_child)
Copy link
Member

@bjester bjester Dec 18, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dead code-- Django's get() should throw ContentNode.DoesNotExist if not found. This is repeated in each test.


# Set a different language on the original node
spanish_language = Language.objects.get(id="es")
contentnode.language = spanish_language
contentnode.save()

sync_channel(
self.derivative_channel,
sync_titles_and_descriptions=False,
sync_resource_details=True,
sync_files=False,
sync_assessment_items=False,
)

self.assertTrue(self.channel.has_changes())
self.assertTrue(self.derivative_channel.has_changes())

target_child.refresh_from_db()
self.assertEqual(target_child.language, spanish_language)

def test_sync_provider_field(self):
"""
Test that the provider field is synced correctly when sync_resource_details is True.
"""
self.assertFalse(self.channel.has_changes())
self.assertFalse(self.derivative_channel.has_changes())

contentnode = (
self.channel.main_tree.get_descendants()
.exclude(kind_id=content_kinds.TOPIC)
.first()
)

target_child = self.derivative_channel.main_tree.get_descendants().get(
source_node_id=contentnode.node_id
)

self.assertIsNotNone(target_child)

# Set a provider on the original node
contentnode.provider = "Test Provider Organization"
contentnode.save()

sync_channel(
self.derivative_channel,
sync_titles_and_descriptions=False,
sync_resource_details=True,
sync_files=False,
sync_assessment_items=False,
)

self.assertTrue(self.channel.has_changes())
self.assertTrue(self.derivative_channel.has_changes())

target_child.refresh_from_db()
self.assertEqual(target_child.provider, "Test Provider Organization")

def test_sync_aggregator_field(self):
"""
Test that the aggregator field is synced correctly when sync_resource_details is True.
"""
self.assertFalse(self.channel.has_changes())
self.assertFalse(self.derivative_channel.has_changes())

contentnode = (
self.channel.main_tree.get_descendants()
.exclude(kind_id=content_kinds.TOPIC)
.first()
)

target_child = self.derivative_channel.main_tree.get_descendants().get(
source_node_id=contentnode.node_id
)

self.assertIsNotNone(target_child)

# Set an aggregator on the original node
contentnode.aggregator = "Test Aggregator Organization"
contentnode.save()

sync_channel(
self.derivative_channel,
sync_titles_and_descriptions=False,
sync_resource_details=True,
sync_files=False,
sync_assessment_items=False,
)

self.assertTrue(self.channel.has_changes())
self.assertTrue(self.derivative_channel.has_changes())

target_child.refresh_from_db()
self.assertEqual(target_child.aggregator, "Test Aggregator Organization")

def test_sync_role_visibility_field(self):
"""
Test that the role_visibility field is synced correctly when sync_resource_details is True.
"""
self.assertFalse(self.channel.has_changes())
self.assertFalse(self.derivative_channel.has_changes())

contentnode = (
self.channel.main_tree.get_descendants()
.exclude(kind_id=content_kinds.TOPIC)
.first()
)

target_child = self.derivative_channel.main_tree.get_descendants().get(
source_node_id=contentnode.node_id
)

self.assertIsNotNone(target_child)

# Set role_visibility to COACH on the original node
contentnode.role_visibility = roles.COACH
contentnode.save()

sync_channel(
self.derivative_channel,
sync_titles_and_descriptions=False,
sync_resource_details=True,
sync_files=False,
sync_assessment_items=False,
)

self.assertTrue(self.channel.has_changes())
self.assertTrue(self.derivative_channel.has_changes())

target_child.refresh_from_db()
self.assertEqual(target_child.role_visibility, roles.COACH)

def test_sync_all_missing_fields(self):
"""
Test that all four previously missing fields (language, provider, aggregator,
role_visibility) are synced together when sync_resource_details is True.
"""
self.assertFalse(self.channel.has_changes())
self.assertFalse(self.derivative_channel.has_changes())

contentnode = (
self.channel.main_tree.get_descendants()
.exclude(kind_id=content_kinds.TOPIC)
.first()
)

target_child = self.derivative_channel.main_tree.get_descendants().get(
source_node_id=contentnode.node_id
)

self.assertIsNotNone(target_child)

# Set all four fields on the original node
french_language = Language.objects.get(id="fr")
contentnode.language = french_language
contentnode.provider = "Comprehensive Test Provider"
contentnode.aggregator = "Comprehensive Test Aggregator"
contentnode.role_visibility = roles.COACH
contentnode.save()

sync_channel(
self.derivative_channel,
sync_titles_and_descriptions=False,
sync_resource_details=True,
sync_files=False,
sync_assessment_items=False,
)

self.assertTrue(self.channel.has_changes())
self.assertTrue(self.derivative_channel.has_changes())

target_child.refresh_from_db()
self.assertEqual(target_child.language, french_language)
self.assertEqual(target_child.provider, "Comprehensive Test Provider")
self.assertEqual(target_child.aggregator, "Comprehensive Test Aggregator")
self.assertEqual(target_child.role_visibility, roles.COACH)

def test_sync_missing_fields_not_synced_without_flag(self):
"""
Test that the four fields (language, provider, aggregator, role_visibility)
are NOT synced when sync_resource_details is False.
"""
self.assertFalse(self.channel.has_changes())
self.assertFalse(self.derivative_channel.has_changes())

contentnode = (
self.channel.main_tree.get_descendants()
.exclude(kind_id=content_kinds.TOPIC)
.first()
)

target_child = self.derivative_channel.main_tree.get_descendants().get(
source_node_id=contentnode.node_id
)

self.assertIsNotNone(target_child)

# Store original values
original_language = target_child.language
original_provider = target_child.provider
original_aggregator = target_child.aggregator
original_role_visibility = target_child.role_visibility

# Modify all four fields in the original node
german_language = Language.objects.get(id="de")
contentnode.language = german_language
contentnode.provider = "Should Not Sync Provider"
contentnode.aggregator = "Should Not Sync Aggregator"
contentnode.role_visibility = roles.COACH
contentnode.save()

# Sync WITHOUT sync_resource_details
sync_channel(
self.derivative_channel,
sync_titles_and_descriptions=False,
sync_resource_details=False,
sync_files=False,
sync_assessment_items=False,
)

target_child.refresh_from_db()

# Verify fields remain unchanged
self.assertEqual(target_child.language, original_language)
self.assertEqual(target_child.provider, original_provider)
self.assertEqual(target_child.aggregator, original_aggregator)
self.assertEqual(target_child.role_visibility, original_role_visibility)


class ContentIDTestCase(SyncTestMixin, StudioAPITestCase):
def setUp(self):
Expand Down
4 changes: 4 additions & 0 deletions contentcuration/contentcuration/utils/sync.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,10 @@ def sync_node(
"license_description",
"copyright_holder",
"author",
"language",
"provider",
"aggregator",
"role_visibility",
"extra_fields",
"categories",
"learner_needs",
Expand Down