Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of github.com:openedx/edx-platform into HEAD

Browse Source
This commit is contained in:
Taras Lytvynenko
2023-11-15 16:50:51 +02:00
parent 13d7945370 034d6322d0
commit 369903986c
163 changed files with 3298 additions and 3460 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
openedx/core/djangoapps/content_tagging/api.py
View File

@@ -58,6 +58,9 @@ def set_taxonomy_orgs(
If not `all_orgs`, the taxonomy is associated with each org in the `orgs` list. If that list is empty, the
taxonomy is not associated with any orgs.
"""
if taxonomy.system_defined:
raise ValueError("Cannot set orgs for a system-defined taxonomy")
TaxonomyOrg.objects.filter(
taxonomy=taxonomy,
rel_type=relationship,

79
openedx/core/djangoapps/content_tagging/rest_api/v1/serializers.py
View File

@@ -2,22 +2,99 @@
API Serializers for content tagging org
"""
from __future__ import annotations
from django.core.exceptions import ObjectDoesNotExist
from rest_framework import serializers, fields
from openedx_tagging.core.tagging.rest_api.v1.serializers import (
TaxonomyListQueryParamsSerializer,
TaxonomySerializer,
)
from organizations.models import Organization
class OptionalSlugRelatedField(serializers.SlugRelatedField):
"""
Modifies the DRF serializer SlugRelatedField.
Non-existent slug values are represented internally as an empty queryset, instead of throwing a validation error.
"""
def to_internal_value(self, data):
"""
Returns the object related to the given slug value, or an empty queryset if not found.
"""
queryset = self.get_queryset()
try:
return queryset.get(**{self.slug_field: data})
except ObjectDoesNotExist:
return queryset.none()
except (TypeError, ValueError):
self.fail('invalid')
class TaxonomyOrgListQueryParamsSerializer(TaxonomyListQueryParamsSerializer):
"""
Serializer for the query params for the GET view
"""
org: fields.Field = serializers.SlugRelatedField(
org: fields.Field = OptionalSlugRelatedField(
slug_field="short_name",
queryset=Organization.objects.all(),
required=False,
)
class TaxonomyUpdateOrgBodySerializer(serializers.Serializer):
"""
Serializer for the body params for the update orgs action
"""
orgs: fields.Field = serializers.SlugRelatedField(
many=True,
slug_field="short_name",
queryset=Organization.objects.all(),
required=False,
)
all_orgs: fields.Field = serializers.BooleanField(required=False)
def validate(self, attrs: dict) -> dict:
"""
Validate the serializer data
"""
if bool(attrs.get("orgs") is not None) == bool(attrs.get("all_orgs")):
raise serializers.ValidationError(
"You must specify either orgs or all_orgs, but not both."
)
return attrs
class TaxonomyOrgSerializer(TaxonomySerializer):
"""
Serializer for Taxonomy objects inclusing the associated orgs
"""
orgs = serializers.SerializerMethodField()
all_orgs = serializers.SerializerMethodField()
def get_orgs(self, obj) -> list[str]:
"""
Return the list of orgs for the taxonomy.
"""
return [taxonomy_org.org.short_name for taxonomy_org in obj.taxonomyorg_set.all() if taxonomy_org.org]
def get_all_orgs(self, obj) -> bool:
"""
Return True if the taxonomy is associated with all orgs.
"""
return obj.taxonomyorg_set.filter(org__isnull=True).exists()
class Meta:
model = TaxonomySerializer.Meta.model
fields = TaxonomySerializer.Meta.fields + ["orgs", "all_orgs"]
read_only_fields = ["orgs", "all_orgs"]

699
openedx/core/djangoapps/content_tagging/rest_api/v1/tests/test_views.py
View File

@@ -5,10 +5,12 @@ Tests tagging rest api views
from __future__ import annotations
from urllib.parse import parse_qs, urlparse
import json
import abc
import ddt
from django.contrib.auth import get_user_model
from django.core.files.uploadedfile import SimpleUploadedFile
from opaque_keys.edx.locator import BlockUsageLocator, CourseLocator
from openedx_tagging.core.tagging.models import Tag, Taxonomy
from openedx_tagging.core.tagging.models.system_defined import SystemDefinedTaxonomy
@@ -40,8 +42,12 @@ User = get_user_model()
TAXONOMY_ORG_LIST_URL = "/api/content_tagging/v1/taxonomies/"
TAXONOMY_ORG_DETAIL_URL = "/api/content_tagging/v1/taxonomies/{pk}/"
TAXONOMY_ORG_UPDATE_ORG_URL = "/api/content_tagging/v1/taxonomies/{pk}/orgs/"
OBJECT_TAG_UPDATE_URL = "/api/content_tagging/v1/object_tags/{object_id}/?taxonomy={taxonomy_id}"
TAXONOMY_TEMPLATE_URL = "/api/content_tagging/v1/taxonomies/import/{filename}"
TAXONOMY_CREATE_IMPORT_URL = "/api/content_tagging/v1/taxonomies/import/"
TAXONOMY_TAGS_IMPORT_URL = "/api/content_tagging/v1/taxonomies/{pk}/tags/import/"
TAXONOMY_TAGS_URL = "/api/content_tagging/v1/taxonomies/{pk}/tags/"
def check_taxonomy(
@@ -342,6 +348,8 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
("orgA", ["st1", "st2", "t1", "t2", "tA1", "tA2", "tBA1", "tBA2"]),
("orgB", ["st1", "st2", "t1", "t2", "tB1", "tB2", "tBA1", "tBA2"]),
("orgX", ["st1", "st2", "t1", "t2"]),
# Non-existent orgs are ignored
("invalidOrg", ["st1", "st2", "t1", "t2"]),
)
@ddt.unpack
def test_list_taxonomy_org_filter(self, org_parameter: str, expected_taxonomies: list[str]) -> None:
@@ -354,20 +362,6 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
expected_taxonomies=expected_taxonomies,
)
def test_list_taxonomy_invalid_org(self) -> None:
"""
Tests that using an invalid org in the filter will raise BAD_REQUEST
"""
url = TAXONOMY_ORG_LIST_URL
self.client.force_authenticate(user=self.staff)
query_params = {"org": "invalidOrg"}
response = self.client.get(url, query_params, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST
@ddt.data(
("user", (), None),
("staffA", ["tA2", "tBA1", "tBA2"], None),
@@ -454,7 +448,7 @@ class TestTaxonomyListCreateViewSet(TestTaxonomyObjectsMixin, APITestCase):
# Also checks if the taxonomy was associated with the org
if user_attr == "staffA":
assert TaxonomyOrg.objects.filter(taxonomy=response.data["id"], org=self.orgA).exists()
assert response.data["orgs"] == [self.orgA.short_name]
@ddt.ddt
@@ -1044,6 +1038,208 @@ class TestTaxonomyDeleteViewSet(TestTaxonomyChangeMixin, APITestCase):
assert response.status_code == status.HTTP_404_NOT_FOUND
@skip_unless_cms
@ddt.ddt
class TestTaxonomyUpdateOrg(TestTaxonomyObjectsMixin, APITestCase):
"""
Test cases for updating orgs from taxonomies
"""
def test_update_org(self) -> None:
"""
Tests that taxonomy admin can add/remove orgs from a taxonomy
"""
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
self.client.force_authenticate(user=self.staff)
response = self.client.put(url, {"orgs": [self.orgB.short_name, self.orgX.short_name]}, format="json")
assert response.status_code == status.HTTP_200_OK
# Check that the orgs were updated
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
response = self.client.get(url)
assert response.data["orgs"] == [self.orgB.short_name, self.orgX.short_name]
assert not response.data["all_orgs"]
def test_update_all_org(self) -> None:
"""
Tests that taxonomy admin can associate a taxonomy to all orgs
"""
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
self.client.force_authenticate(user=self.staff)
response = self.client.put(url, {"all_orgs": True}, format="json")
assert response.status_code == status.HTTP_200_OK
# Check that the orgs were updated
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
response = self.client.get(url)
assert response.data["orgs"] == []
assert response.data["all_orgs"]
def test_update_no_org(self) -> None:
"""
Tests that taxonomy admin can associate a taxonomy no orgs
"""
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
self.client.force_authenticate(user=self.staff)
response = self.client.put(url, {"orgs": []}, format="json")
assert response.status_code == status.HTTP_200_OK
# Check that the orgs were updated
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
response = self.client.get(url)
assert response.data["orgs"] == []
assert not response.data["all_orgs"]
@ddt.data(
(True, ["orgX"], "Using both all_orgs and orgs parameters should throw error"),
(False, None, "Using neither all_orgs or orgs parameter should throw error"),
(None, None, "Using neither all_orgs or orgs parameter should throw error"),
(False, 'InvalidOrg', "Passing an invalid org should throw error"),
)
@ddt.unpack
def test_update_org_invalid_inputs(self, all_orgs: bool, orgs: list[str], reason: str) -> None:
"""
Tests if passing both or none of all_orgs and orgs parameters throws error
"""
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
self.client.force_authenticate(user=self.staff)
# Set body cleaning empty values
body = {k: v for k, v in {"all_orgs": all_orgs, "orgs": orgs}.items() if v is not None}
response = self.client.put(url, body, format="json")
assert response.status_code == status.HTTP_400_BAD_REQUEST, reason
# Check that the orgs didn't change
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
response = self.client.get(url)
assert response.data["orgs"] == [self.orgA.short_name]
def test_update_org_system_defined(self) -> None:
"""
Tests that is not possible to change the orgs associated with a system defined taxonomy
"""
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.st1.pk)
self.client.force_authenticate(user=self.staff)
response = self.client.put(url, {"orgs": [self.orgA.short_name]}, format="json")
assert response.status_code in [status.HTTP_403_FORBIDDEN, status.HTTP_400_BAD_REQUEST]
# Check that the orgs didn't change
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.st1.pk)
response = self.client.get(url)
assert response.data["orgs"] == []
assert response.data["all_orgs"]
@ddt.data(
"staffA",
"content_creatorA",
"instructorA",
"library_staffA",
"course_instructorA",
"course_staffA",
"library_userA",
)
def test_update_org_no_perm(self, user_attr: str) -> None:
"""
Tests that only taxonomy admins can associate orgs to taxonomies
"""
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tA1.pk)
user = getattr(self, user_attr)
self.client.force_authenticate(user=user)
response = self.client.put(url, {"orgs": []}, format="json")
assert response.status_code == status.HTTP_403_FORBIDDEN
# Check that the orgs didn't change
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
response = self.client.get(url)
assert response.data["orgs"] == [self.orgA.short_name]
def test_update_org_check_permissions_orgA(self) -> None:
"""
Tests that adding an org to a taxonomy allow org level admins to edit it
"""
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staffA)
response = self.client.put(url, {"name": "new name"}, format="json")
# User staffA can't update metadata from a taxonomy from orgB
assert response.status_code == status.HTTP_404_NOT_FOUND
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staff)
# Add the taxonomy tB1 to orgA
response = self.client.put(url, {"orgs": [self.orgA.short_name]}, format="json")
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staffA)
response = self.client.put(url, {"name": "new name"}, format="json")
# Now staffA can change the metadata from a tB1 because it's associated with orgA
assert response.status_code == status.HTTP_200_OK
def test_update_org_check_permissions_all_orgs(self) -> None:
"""
Tests that adding an org to all orgs only let taxonomy global admins to edit it
"""
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
self.client.force_authenticate(user=self.staffA)
response = self.client.put(url, {"name": "new name"}, format="json")
# User staffA can update metadata from a taxonomy from orgA
assert response.status_code == status.HTTP_200_OK
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staff)
# Add the taxonomy tA1 to all orgs
response = self.client.put(url, {"all_orgs": True}, format="json")
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staffA)
response = self.client.put(url, {"name": "new name"}, format="json")
# Now staffA can't change the metadata from a tA1 because only global taxonomy admins can edit all orgs
# taxonomies
assert response.status_code == status.HTTP_403_FORBIDDEN
def test_update_org_check_permissions_no_orgs(self) -> None:
"""
Tests that remove all orgs from a taxonomy only let taxonomy global admins to edit it
"""
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tA1.pk)
self.client.force_authenticate(user=self.staffA)
response = self.client.put(url, {"name": "new name"}, format="json")
# User staffA can update metadata from a taxonomy from orgA
assert response.status_code == status.HTTP_200_OK
url = TAXONOMY_ORG_UPDATE_ORG_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staff)
# Remove all orgs from tA1
response = self.client.put(url, {"orgs": []}, format="json")
url = TAXONOMY_ORG_DETAIL_URL.format(pk=self.tB1.pk)
self.client.force_authenticate(user=self.staffA)
response = self.client.put(url, {"name": "new name"}, format="json")
# Now staffA can't change the metadata from a tA1 because only global taxonomy admins can edit no orgs
# taxonomies
assert response.status_code == status.HTTP_404_NOT_FOUND
class TestObjectTagMixin(TestTaxonomyObjectsMixin):
"""
Sets up data for testing ObjectTags.
@@ -1136,13 +1332,19 @@ class TestObjectTagViewSet(TestObjectTagMixin, APITestCase):
assert response.status_code == expected_status
if status.is_success(expected_status):
assert len(response.data) == len(tag_values)
assert set(t["value"] for t in response.data) == set(tag_values)
tags_by_taxonomy = response.data[str(self.courseA)]["taxonomies"]
if tag_values:
response_taxonomy = tags_by_taxonomy[0]
assert response_taxonomy["name"] == taxonomy.name
response_tags = response_taxonomy["tags"]
assert [t["value"] for t in response_tags] == tag_values
else:
assert tags_by_taxonomy == [] # No tags are set from any taxonomy
# Check that re-fetching the tags returns what we set
response = self.client.get(url, format="json")
assert status.is_success(response.status_code)
assert set(t["value"] for t in response.data) == set(tag_values)
new_response = self.client.get(url, format="json")
assert status.is_success(new_response.status_code)
assert new_response.data == response.data
@ddt.data(
"staffA",
@@ -1214,13 +1416,19 @@ class TestObjectTagViewSet(TestObjectTagMixin, APITestCase):
assert response.status_code == expected_status
if status.is_success(expected_status):
assert len(response.data) == len(tag_values)
assert set(t["value"] for t in response.data) == set(tag_values)
tags_by_taxonomy = response.data[str(self.xblockA)]["taxonomies"]
if tag_values:
response_taxonomy = tags_by_taxonomy[0]
assert response_taxonomy["name"] == taxonomy.name
response_tags = response_taxonomy["tags"]
assert [t["value"] for t in response_tags] == tag_values
else:
assert tags_by_taxonomy == [] # No tags are set from any taxonomy
# Check that re-fetching the tags returns what we set
response = self.client.get(url, format="json")
assert status.is_success(response.status_code)
assert set(t["value"] for t in response.data) == set(tag_values)
new_response = self.client.get(url, format="json")
assert status.is_success(new_response.status_code)
assert new_response.data == response.data
@ddt.data(
"staffA",
@@ -1324,3 +1532,444 @@ class TestDownloadTemplateView(APITestCase):
url = TAXONOMY_TEMPLATE_URL.format(filename="template.txt")
response = self.client.post(url)
assert response.status_code == status.HTTP_405_METHOD_NOT_ALLOWED
class ImportTaxonomyMixin(TestTaxonomyObjectsMixin):
"""
Mixin to test importing taxonomies.
"""
def _get_file(self, tags: list, file_format: str) -> SimpleUploadedFile:
"""
Returns a file for the given format.
"""
if file_format == "csv":
csv_data = "id,value"
for tag in tags:
csv_data += f"\n{tag['id']},{tag['value']}"
return SimpleUploadedFile("taxonomy.csv", csv_data.encode(), content_type="text/csv")
else: # json
json_data = {"tags": tags}
return SimpleUploadedFile("taxonomy.json", json.dumps(json_data).encode(), content_type="application/json")
@skip_unless_cms
@ddt.ddt
class TestCreateImportView(ImportTaxonomyMixin, APITestCase):
"""
Tests the create/import taxonomy action.
"""
@ddt.data(
"csv",
"json",
)
def test_import_global_admin(self, file_format: str) -> None:
"""
Tests importing a valid taxonomy file with a global admin.
"""
url = TAXONOMY_CREATE_IMPORT_URL
new_tags = [
{"id": "tag_1", "value": "Tag 1"},
{"id": "tag_2", "value": "Tag 2"},
{"id": "tag_3", "value": "Tag 3"},
{"id": "tag_4", "value": "Tag 4"},
]
file = self._get_file(new_tags, file_format)
self.client.force_authenticate(user=self.staff)
response = self.client.post(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_201_CREATED
# Check if the taxonomy was created
taxonomy = response.data
assert taxonomy["name"] == "Imported Taxonomy name"
assert taxonomy["description"] == "Imported Taxonomy description"
# Check if the tags were created
url = TAXONOMY_TAGS_URL.format(pk=taxonomy["id"])
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == len(new_tags)
for i, tag in enumerate(tags):
assert tag["value"] == new_tags[i]["value"]
# Check if the taxonomy was no association with orgs
assert len(taxonomy["orgs"]) == 0
@ddt.data(
"csv",
"json",
)
def test_import_orgA_admin(self, file_format: str) -> None:
"""
Tests importing a valid taxonomy file with a orgA admin.
"""
url = TAXONOMY_CREATE_IMPORT_URL
new_tags = [
{"id": "tag_1", "value": "Tag 1"},
{"id": "tag_2", "value": "Tag 2"},
{"id": "tag_3", "value": "Tag 3"},
{"id": "tag_4", "value": "Tag 4"},
]
file = self._get_file(new_tags, file_format)
self.client.force_authenticate(user=self.staffA)
response = self.client.post(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_201_CREATED
# Check if the taxonomy was created
taxonomy = response.data
assert taxonomy["name"] == "Imported Taxonomy name"
assert taxonomy["description"] == "Imported Taxonomy description"
# Check if the tags were created
url = TAXONOMY_TAGS_URL.format(pk=taxonomy["id"])
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == len(new_tags)
for i, tag in enumerate(tags):
assert tag["value"] == new_tags[i]["value"]
# Check if the taxonomy was associated with the orgA
assert len(taxonomy["orgs"]) == 1
assert taxonomy["orgs"][0] == self.orgA.short_name
def test_import_no_file(self) -> None:
"""
Tests importing a taxonomy without a file.
"""
url = TAXONOMY_CREATE_IMPORT_URL
self.client.force_authenticate(user=self.staff)
response = self.client.post(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.data["file"][0] == "No file was submitted."
# Check if the taxonomy was not created
assert not Taxonomy.objects.filter(name="Imported Taxonomy name").exists()
@ddt.data(
"csv",
"json",
)
def test_import_no_name(self, file_format) -> None:
"""
Tests importing a taxonomy without specifing a name.
"""
url = TAXONOMY_CREATE_IMPORT_URL
file = SimpleUploadedFile(f"taxonomy.{file_format}", b"invalid file content")
self.client.force_authenticate(user=self.staff)
response = self.client.post(
url,
{
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.data["taxonomy_name"][0] == "This field is required."
# Check if the taxonomy was not created
assert not Taxonomy.objects.filter(name="Imported Taxonomy name").exists()
def test_import_invalid_format(self) -> None:
"""
Tests importing a taxonomy with an invalid file format.
"""
url = TAXONOMY_CREATE_IMPORT_URL
file = SimpleUploadedFile("taxonomy.invalid", b"invalid file content")
self.client.force_authenticate(user=self.staff)
response = self.client.post(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.data["file"][0] == "File type not supported: invalid"
# Check if the taxonomy was not created
assert not Taxonomy.objects.filter(name="Imported Taxonomy name").exists()
@ddt.data(
"csv",
"json",
)
def test_import_invalid_content(self, file_format) -> None:
"""
Tests importing a taxonomy with an invalid file content.
"""
url = TAXONOMY_CREATE_IMPORT_URL
file = SimpleUploadedFile(f"taxonomy.{file_format}", b"invalid file content")
self.client.force_authenticate(user=self.staff)
response = self.client.post(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert f"Invalid '.{file_format}' format:" in response.data
# Check if the taxonomy was not created
assert not Taxonomy.objects.filter(name="Imported Taxonomy name").exists()
def test_import_no_perm(self) -> None:
"""
Tests importing a taxonomy using a user without permission.
"""
url = TAXONOMY_CREATE_IMPORT_URL
new_tags = [
{"id": "tag_1", "value": "Tag 1"},
{"id": "tag_2", "value": "Tag 2"},
{"id": "tag_3", "value": "Tag 3"},
{"id": "tag_4", "value": "Tag 4"},
]
file = self._get_file(new_tags, "json")
self.client.force_authenticate(user=self.user)
response = self.client.post(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_403_FORBIDDEN
# Check if the taxonomy was not created
assert not Taxonomy.objects.filter(name="Imported Taxonomy name").exists()
@skip_unless_cms
@ddt.ddt
class TestImportTagsView(ImportTaxonomyMixin, APITestCase):
"""
Tests the taxonomy import tags action.
"""
def setUp(self):
ImportTaxonomyMixin.setUp(self)
self.taxonomy = Taxonomy.objects.create(
name="Test import taxonomy",
)
tag_1 = Tag.objects.create(
taxonomy=self.taxonomy,
external_id="old_tag_1",
value="Old tag 1",
)
tag_2 = Tag.objects.create(
taxonomy=self.taxonomy,
external_id="old_tag_2",
value="Old tag 2",
)
self.old_tags = [tag_1, tag_2]
@ddt.data(
"csv",
"json",
)
def test_import(self, file_format: str) -> None:
"""
Tests importing a valid taxonomy file.
"""
url = TAXONOMY_TAGS_IMPORT_URL.format(pk=self.taxonomy.id)
new_tags = [
{"id": "tag_1", "value": "Tag 1"},
{"id": "tag_2", "value": "Tag 2"},
{"id": "tag_3", "value": "Tag 3"},
{"id": "tag_4", "value": "Tag 4"},
]
file = self._get_file(new_tags, file_format)
self.client.force_authenticate(user=self.staff)
response = self.client.put(
url,
{"file": file},
format="multipart"
)
assert response.status_code == status.HTTP_200_OK
# Check if the tags were created
url = TAXONOMY_TAGS_URL.format(pk=self.taxonomy.id)
response = self.client.get(url)
tags = response.data["results"]
all_tags = [{"value": tag.value} for tag in self.old_tags] + new_tags
assert len(tags) == len(all_tags)
for i, tag in enumerate(tags):
assert tag["value"] == all_tags[i]["value"]
def test_import_no_file(self) -> None:
"""
Tests importing a taxonomy without a file.
"""
url = TAXONOMY_TAGS_IMPORT_URL.format(pk=self.taxonomy.id)
self.client.force_authenticate(user=self.staff)
response = self.client.put(
url,
{},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.data["file"][0] == "No file was submitted."
# Check if the taxonomy was not changed
url = TAXONOMY_TAGS_URL.format(pk=self.taxonomy.id)
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == len(self.old_tags)
for i, tag in enumerate(tags):
assert tag["value"] == self.old_tags[i].value
def test_import_invalid_format(self) -> None:
"""
Tests importing a taxonomy with an invalid file format.
"""
url = TAXONOMY_TAGS_IMPORT_URL.format(pk=self.taxonomy.id)
file = SimpleUploadedFile("taxonomy.invalid", b"invalid file content")
self.client.force_authenticate(user=self.staff)
response = self.client.put(
url,
{"file": file},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.data["file"][0] == "File type not supported: invalid"
# Check if the taxonomy was not changed
url = TAXONOMY_TAGS_URL.format(pk=self.taxonomy.id)
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == len(self.old_tags)
for i, tag in enumerate(tags):
assert tag["value"] == self.old_tags[i].value
@ddt.data(
"csv",
"json",
)
def test_import_invalid_content(self, file_format) -> None:
"""
Tests importing a taxonomy with an invalid file content.
"""
url = TAXONOMY_TAGS_IMPORT_URL.format(pk=self.taxonomy.id)
file = SimpleUploadedFile(f"taxonomy.{file_format}", b"invalid file content")
self.client.force_authenticate(user=self.staff)
response = self.client.put(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert f"Invalid '.{file_format}' format:" in response.data
# Check if the taxonomy was not changed
url = TAXONOMY_TAGS_URL.format(pk=self.taxonomy.id)
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == len(self.old_tags)
for i, tag in enumerate(tags):
assert tag["value"] == self.old_tags[i].value
@ddt.data(
"csv",
"json",
)
def test_import_free_text(self, file_format) -> None:
"""
Tests importing a taxonomy with an invalid file content.
"""
self.taxonomy.allow_free_text = True
self.taxonomy.save()
url = TAXONOMY_TAGS_IMPORT_URL.format(pk=self.taxonomy.id)
new_tags = [
{"id": "tag_1", "value": "Tag 1"},
{"id": "tag_2", "value": "Tag 2"},
{"id": "tag_3", "value": "Tag 3"},
{"id": "tag_4", "value": "Tag 4"},
]
file = self._get_file(new_tags, file_format)
self.client.force_authenticate(user=self.staff)
response = self.client.put(
url,
{"file": file},
format="multipart"
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert response.data == f"Invalid taxonomy ({self.taxonomy.id}): You cannot import a free-form taxonomy."
# Check if the taxonomy has no tags, since it is free text
url = TAXONOMY_TAGS_URL.format(pk=self.taxonomy.id)
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == 0
def test_import_no_perm(self) -> None:
"""
Tests importing a taxonomy using a user without permission.
"""
url = TAXONOMY_TAGS_IMPORT_URL.format(pk=self.taxonomy.id)
new_tags = [
{"id": "tag_1", "value": "Tag 1"},
{"id": "tag_2", "value": "Tag 2"},
{"id": "tag_3", "value": "Tag 3"},
{"id": "tag_4", "value": "Tag 4"},
]
file = self._get_file(new_tags, "json")
self.client.force_authenticate(user=self.user)
response = self.client.put(
url,
{
"taxonomy_name": "Imported Taxonomy name",
"taxonomy_description": "Imported Taxonomy description",
"file": file,
},
format="multipart"
)
assert response.status_code == status.HTTP_404_NOT_FOUND
# Check if the taxonomy was not changed
url = TAXONOMY_TAGS_URL.format(pk=self.taxonomy.id)
self.client.force_authenticate(user=self.staff)
response = self.client.get(url)
tags = response.data["results"]
assert len(tags) == len(self.old_tags)
for i, tag in enumerate(tags):
assert tag["value"] == self.old_tags[i].value

71
openedx/core/djangoapps/content_tagging/rest_api/v1/views.py
View File

@@ -1,23 +1,29 @@
"""
Tagging Org API Views
"""
from openedx_tagging.core.tagging import rules as oel_tagging_rules
from openedx_tagging.core.tagging.rest_api.v1.views import ObjectTagView, TaxonomyView
from rest_framework import status
from rest_framework.decorators import action
from rest_framework.exceptions import PermissionDenied
from rest_framework.request import Request
from rest_framework.response import Response
from ...api import (
create_taxonomy,
get_taxonomy,
get_taxonomies,
get_taxonomies_for_org,
set_taxonomy_orgs,
)
from ...rules import get_admin_orgs
from .serializers import TaxonomyOrgListQueryParamsSerializer
from .serializers import TaxonomyOrgListQueryParamsSerializer, TaxonomyOrgSerializer, TaxonomyUpdateOrgBodySerializer
from .filters import ObjectTagTaxonomyOrgFilterBackend, UserOrgFilterBackend
class TaxonomyOrgView(TaxonomyView):
"""
View to list, create, retrieve, update, or delete Taxonomies.
View to list, create, retrieve, update, delete, export or import Taxonomies.
This view extends the TaxonomyView to add Organization filters.
Refer to TaxonomyView docstring for usage details.
@@ -36,6 +42,7 @@ class TaxonomyOrgView(TaxonomyView):
"""
filter_backends = [UserOrgFilterBackend]
serializer_class = TaxonomyOrgSerializer
def get_queryset(self):
"""
@@ -48,11 +55,15 @@ class TaxonomyOrgView(TaxonomyView):
query_params = TaxonomyOrgListQueryParamsSerializer(data=self.request.query_params.dict())
query_params.is_valid(raise_exception=True)
enabled = query_params.validated_data.get("enabled", None)
# If org filtering was requested, then use it, even if the org is invalid/None
org = query_params.validated_data.get("org", None)
if org:
return get_taxonomies_for_org(enabled, org)
if "org" in query_params.validated_data:
queryset = get_taxonomies_for_org(enabled, org)
else:
return get_taxonomies(enabled)
queryset = get_taxonomies(enabled)
return queryset.prefetch_related("taxonomyorg_set")
def perform_create(self, serializer):
"""
@@ -61,6 +72,52 @@ class TaxonomyOrgView(TaxonomyView):
user_admin_orgs = get_admin_orgs(self.request.user)
serializer.instance = create_taxonomy(**serializer.validated_data, orgs=user_admin_orgs)
@action(detail=False, url_path="import", methods=["post"])
def create_import(self, request: Request, **kwargs) -> Response:
"""
Creates a new taxonomy with the given orgs and imports the tags from the uploaded file.
"""
response = super().create_import(request, **kwargs)
# If creation was successful, set the orgs for the new taxonomy
if status.is_success(response.status_code):
# ToDo: This code is temporary
# In the future, the orgs parameter will be defined in the request body from the frontend
# See: https://github.com/openedx/modular-learning/issues/116
if oel_tagging_rules.is_taxonomy_admin(request.user):
orgs = None
else:
orgs = get_admin_orgs(request.user)
taxonomy = get_taxonomy(response.data["id"])
assert taxonomy
set_taxonomy_orgs(taxonomy, all_orgs=False, orgs=orgs)
serializer = self.get_serializer(taxonomy)
return Response(serializer.data, status=status.HTTP_201_CREATED)
return response
@action(detail=True, methods=["put"])
def orgs(self, request, **_kwargs) -> Response:
"""
Update the orgs associated with taxonomies.
"""
taxonomy = self.get_object()
perm = "oel_tagging.update_orgs"
if not request.user.has_perm(perm, taxonomy):
raise PermissionDenied("You do not have permission to update the orgs associated with this taxonomy.")
body = TaxonomyUpdateOrgBodySerializer(
data=request.data,
)
body.is_valid(raise_exception=True)
orgs = body.validated_data.get("orgs")
all_orgs: bool = body.validated_data.get("all_orgs", False)
set_taxonomy_orgs(taxonomy=taxonomy, all_orgs=all_orgs, orgs=orgs)
return Response()
class ObjectTagOrgView(ObjectTagView):
"""

2
openedx/core/djangoapps/content_tagging/rules.py
View File

@@ -281,7 +281,7 @@ rules.set_perm("oel_tagging.add_taxonomy", can_create_taxonomy)
rules.set_perm("oel_tagging.change_taxonomy", can_change_taxonomy)
rules.set_perm("oel_tagging.delete_taxonomy", can_change_taxonomy)
rules.set_perm("oel_tagging.view_taxonomy", can_view_taxonomy)
rules.set_perm("oel_tagging.export_taxonomy", can_view_taxonomy)
rules.add_perm("oel_tagging.update_orgs", oel_tagging.is_taxonomy_admin)
# Tag
rules.set_perm("oel_tagging.add_tag", can_change_taxonomy_tag)

20
openedx/core/djangoapps/contentserver/middleware.py
View File

@@ -100,17 +100,17 @@ class StaticContentServer(MiddlewareMixin):
safe_course_key = safe_course_key.replace(run='only')
if newrelic:
newrelic.agent.add_custom_parameter('course_id', safe_course_key)
newrelic.agent.add_custom_parameter('org', loc.org)
newrelic.agent.add_custom_parameter('contentserver.path', loc.path)
newrelic.agent.add_custom_attribute('course_id', safe_course_key)
newrelic.agent.add_custom_attribute('org', loc.org)
newrelic.agent.add_custom_attribute('contentserver.path', loc.path)
# Figure out if this is a CDN using us as the origin.
is_from_cdn = StaticContentServer.is_cdn_request(request)
newrelic.agent.add_custom_parameter('contentserver.from_cdn', is_from_cdn)
newrelic.agent.add_custom_attribute('contentserver.from_cdn', is_from_cdn)
# Check if this content is locked or not.
locked = self.is_content_locked(content)
newrelic.agent.add_custom_parameter('contentserver.locked', locked)
newrelic.agent.add_custom_attribute('contentserver.locked', locked)
# Check that user has access to the content.
if not self.is_user_authorized(request, content, loc):
@@ -169,7 +169,7 @@ class StaticContentServer(MiddlewareMixin):
response.status_code = 206 # Partial Content
if newrelic:
newrelic.agent.add_custom_parameter('contentserver.ranged', True)
newrelic.agent.add_custom_attribute('contentserver.ranged', True)
else:
log.warning(
"Cannot satisfy ranges in Range header: %s for content: %s",
@@ -183,8 +183,8 @@ class StaticContentServer(MiddlewareMixin):
response['Content-Length'] = content.length
if newrelic:
newrelic.agent.add_custom_parameter('contentserver.content_len', content.length)
newrelic.agent.add_custom_parameter('contentserver.content_type', content.content_type)
newrelic.agent.add_custom_attribute('contentserver.content_len', content.length)
newrelic.agent.add_custom_attribute('contentserver.content_type', content.content_type)
# "Accept-Ranges: bytes" tells the user that only "bytes" ranges are allowed
response['Accept-Ranges'] = 'bytes'
@@ -214,13 +214,13 @@ class StaticContentServer(MiddlewareMixin):
cache_ttl = CourseAssetCacheTtlConfig.get_cache_ttl()
if cache_ttl > 0 and not is_locked:
if newrelic:
newrelic.agent.add_custom_parameter('contentserver.cacheable', True)
newrelic.agent.add_custom_attribute('contentserver.cacheable', True)
response['Expires'] = StaticContentServer.get_expiration_value(datetime.datetime.utcnow(), cache_ttl)
response['Cache-Control'] = "public, max-age={ttl}, s-maxage={ttl}".format(ttl=cache_ttl)
elif is_locked:
if newrelic:
newrelic.agent.add_custom_parameter('contentserver.cacheable', False)
newrelic.agent.add_custom_attribute('contentserver.cacheable', False)
response['Cache-Control'] = "private, no-cache, no-store"

20
openedx/core/djangoapps/course_apps/toggles.py
View File

@@ -6,19 +6,6 @@ from openedx.core.djangoapps.waffle_utils import CourseWaffleFlag
#: Namespace for use by course apps for creating availability toggles
COURSE_APPS_WAFFLE_NAMESPACE = 'course_apps'
# .. toggle_name: course_apps.proctoring_settings_modal_view
# .. toggle_use_cases: temporary
# .. toggle_implementation: CourseWaffleFlag
# .. toggle_default: False
# .. toggle_description: When enabled, users will be directed to a new proctoring settings
# modal on the Pages and Resources view when accessing proctored exam settings.
# .. toggle_warning: None
# .. toggle_creation_date: 2021-08-17
# .. toggle_target_removal_date: None
PROCTORING_SETTINGS_MODAL_VIEW = CourseWaffleFlag(
f'{COURSE_APPS_WAFFLE_NAMESPACE}.proctoring_settings_modal_view', __name__
)
# .. toggle_name: course_apps.exams_ida
# .. toggle_implementation: CourseWaffleFlag
# .. toggle_default: False
@@ -32,13 +19,6 @@ EXAMS_IDA = CourseWaffleFlag(
)
def proctoring_settings_modal_view_enabled(course_key):
"""
Returns a boolean if proctoring settings modal view is enabled for a course.
"""
return PROCTORING_SETTINGS_MODAL_VIEW.is_enabled(course_key)
def exams_ida_enabled(course_key):
"""
Returns a boolean if exams ida view is enabled for a course.

2
openedx/core/djangoapps/embargo/tests/test_views.py
View File

@@ -148,7 +148,7 @@ class CheckCourseAccessViewTest(CourseApiFactoryMixin, ModuleStoreTestCase):
def test_course_access_endpoint_with_logged_out_user(self):
self.client.logout()
response = self.client.get(self.url, data=self.request_data)
assert response.status_code == 403
assert response.status_code == 401
def test_course_access_endpoint_with_non_staff_user(self):
user = UserFactory(is_staff=False)

10
openedx/core/djangoapps/notifications/config/waffle.py
View File

@@ -37,3 +37,13 @@ SHOW_NOTIFICATIONS_TRAY = CourseWaffleFlag(f"{WAFFLE_NAMESPACE}.show_notificatio
# .. toggle_target_removal_date: 2024-06-01
# .. toggle_tickets: INF-902
ENABLE_NOTIFICATIONS_FILTERS = CourseWaffleFlag(f"{WAFFLE_NAMESPACE}.enable_notifications_filters", __name__)
# .. toggle_name: notifications.enable_coursewide_notifications
# .. toggle_implementation: CourseWaffleFlag
# .. toggle_default: False
# .. toggle_description: Waffle flag to enable coursewide notifications
# .. toggle_use_cases: temporary, open_edx
# .. toggle_creation_date: 2023-10-25
# .. toggle_target_removal_date: 2024-06-01
# .. toggle_tickets: INF-1145
ENABLE_COURSEWIDE_NOTIFICATIONS = CourseWaffleFlag(f"{WAFFLE_NAMESPACE}.enable_coursewide_notifications", __name__)

42
openedx/core/djangoapps/notifications/tasks.py
View File

@@ -101,7 +101,13 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
sender_id = context.pop('sender_id', None)
default_web_config = get_default_values_of_preference(app_name, notification_type).get('web', False)
generated_notification_audience = []
for batch_user_ids in get_list_in_batches(user_ids, batch_size):
if ENABLE_NOTIFICATIONS_FILTERS.is_enabled(course_key):
logger.info(f'Sending notifications to {len(batch_user_ids)} users in {course_key}')
batch_user_ids = NotificationFilter().apply_filters(batch_user_ids, course_key, notification_type)
logger.info(f'After applying filters, sending notifications to {len(batch_user_ids)} users in {course_key}')
# check if what is preferences of user and make decision to send notification or not
preferences = CourseNotificationPreference.objects.filter(
user_id__in=batch_user_ids,
@@ -115,32 +121,26 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
if not preferences:
continue
notifications = []
for preference in preferences:
preference = update_user_preference(preference, preference.user_id, course_key)
if not (
user_id = preference.user_id
preference = update_user_preference(preference, user_id, course_key)
if (
preference and
preference.get_web_config(app_name, notification_type) and
preference.get_app_config(app_name).get('enabled', False)
):
batch_user_ids.remove(preference.user_id)
if ENABLE_NOTIFICATIONS_FILTERS.is_enabled(course_key):
logger.info(f'Sending notifications to {len(batch_user_ids)} users.')
batch_user_ids = NotificationFilter().apply_filters(batch_user_ids, course_key, notification_type)
logger.info(f'After applying filters, sending notifications to {len(batch_user_ids)} users.')
notifications = []
for user_id in batch_user_ids:
notifications.append(
Notification(
user_id=user_id,
app_name=app_name,
notification_type=notification_type,
content_context=context,
content_url=content_url,
course_id=course_key,
notifications.append(
Notification(
user_id=user_id,
app_name=app_name,
notification_type=notification_type,
content_context=context,
content_url=content_url,
course_id=course_key,
)
)
)
generated_notification_audience.append(user_id)
generated_notification_audience.append(user_id)
# send notification to users but use bulk_create
notification_objects = Notification.objects.bulk_create(notifications)
@@ -149,6 +149,8 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
notification_content = notification_objects[0].content
if notifications_generated:
logger.info(f'Temp: Notifications generated for {len(generated_notification_audience)} out of '
f'{len(user_ids)} users - {app_name} - {notification_type} - {course_key}.')
notification_generated_event(
generated_notification_audience, app_name, notification_type, course_key, content_url,
notification_content, sender_id=sender_id

9
openedx/core/djangoapps/notifications/tests/test_views.py
View File

@@ -18,7 +18,11 @@ from rest_framework.test import APIClient, APITestCase
from common.djangoapps.student.models import CourseEnrollment
from common.djangoapps.student.tests.factories import UserFactory
from openedx.core.djangoapps.content.course_overviews.tests.factories import CourseOverviewFactory
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS, SHOW_NOTIFICATIONS_TRAY
from openedx.core.djangoapps.notifications.config.waffle import (
ENABLE_COURSEWIDE_NOTIFICATIONS,
ENABLE_NOTIFICATIONS,
SHOW_NOTIFICATIONS_TRAY
)
from openedx.core.djangoapps.notifications.models import CourseNotificationPreference, Notification
from openedx.core.djangoapps.notifications.serializers import NotificationCourseEnrollmentSerializer
from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
@@ -255,7 +259,8 @@ class UserNotificationPreferenceAPITest(ModuleStoreTestCase):
Test get user notification preference.
"""
self.client.login(username=self.user.username, password=self.TEST_PASSWORD)
response = self.client.get(self.path)
with override_waffle_flag(ENABLE_COURSEWIDE_NOTIFICATIONS, active=True):
response = self.client.get(self.path)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data, self._expected_api_response())
event_name, event_data = mock_emit.call_args[0]

19
openedx/core/djangoapps/notifications/utils.py
View File

@@ -3,7 +3,7 @@ Utils function for notifications app
"""
from common.djangoapps.student.models import CourseEnrollment
from .config.waffle import SHOW_NOTIFICATIONS_TRAY
from .config.waffle import ENABLE_COURSEWIDE_NOTIFICATIONS, SHOW_NOTIFICATIONS_TRAY
def find_app_in_normalized_apps(app_name, apps_list):
@@ -51,3 +51,20 @@ def get_list_in_batches(input_list, batch_size):
list_length = len(input_list)
for index in range(0, list_length, batch_size):
yield input_list[index: index + batch_size]
def filter_course_wide_preferences(course_key, preferences):
"""
If course wide notifications is disabled for course, it filters course_wide
preferences from response
"""
if ENABLE_COURSEWIDE_NOTIFICATIONS.is_enabled(course_key):
return preferences
course_wide_notification_types = ['new_discussion_post', 'new_question_post']
config = preferences['notification_preference_config']
for app_prefs in config.values():
notification_types = app_prefs['notification_types']
for course_wide_type in course_wide_notification_types:
if course_wide_type in notification_types.keys():
notification_types.pop(course_wide_type)
return preferences

5
openedx/core/djangoapps/notifications/views.py
View File

@@ -36,7 +36,7 @@ from .serializers import (
UserCourseNotificationPreferenceSerializer,
UserNotificationPreferenceUpdateSerializer
)
from .utils import get_show_notifications_tray
from .utils import filter_course_wide_preferences, get_show_notifications_tray
@allow_any_authenticated_user()
@@ -183,7 +183,8 @@ class UserNotificationPreferenceView(APIView):
user_preference = CourseNotificationPreference.get_updated_user_course_preferences(request.user, course_id)
serializer = UserCourseNotificationPreferenceSerializer(user_preference)
notification_preferences_viewed_event(request, course_id)
return Response(serializer.data)
preferences = filter_course_wide_preferences(course_id, serializer.data)
return Response(preferences)
def patch(self, request, course_key_string):
"""

28
openedx/core/djangoapps/user_api/tests/test_views.py
View File

@@ -150,12 +150,12 @@ class RoleTestCase(UserApiTestCase):
self.assertHttpMethodNotAllowed(self.request_with_auth("delete", self.LIST_URI))
def test_list_unauthorized(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
@override_settings(DEBUG=True)
@override_settings(EDX_API_KEY=None)
def test_debug_auth(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
@override_settings(DEBUG=False)
@override_settings(EDX_API_KEY=TEST_API_KEY)
@@ -164,7 +164,7 @@ class RoleTestCase(UserApiTestCase):
self.assertHttpOK(
self.request_with_auth("get", self.LIST_URI,
**self.basic_auth("someuser", "somepass")))
self.assertHttpForbidden(
self.assertHttpNotAuthorized(
self.client.get(self.LIST_URI, **self.basic_auth("someuser", "somepass")))
def test_get_list_nonempty(self):
@@ -236,12 +236,12 @@ class UserViewSetTest(UserApiTestCase):
self.assertHttpMethodNotAllowed(self.request_with_auth("delete", self.LIST_URI))
def test_list_unauthorized(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
@override_settings(DEBUG=True)
@override_settings(EDX_API_KEY=None)
def test_debug_auth(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
@override_settings(DEBUG=False)
@override_settings(EDX_API_KEY=TEST_API_KEY)
@@ -250,7 +250,7 @@ class UserViewSetTest(UserApiTestCase):
self.assertHttpOK(
self.request_with_auth("get", self.LIST_URI,
**self.basic_auth('someuser', 'somepass')))
self.assertHttpForbidden(
self.assertHttpNotAuthorized(
self.client.get(self.LIST_URI, **self.basic_auth('someuser', 'somepass')))
def test_get_list_nonempty(self):
@@ -303,7 +303,7 @@ class UserViewSetTest(UserApiTestCase):
self.assertHttpMethodNotAllowed(self.request_with_auth("delete", self.detail_uri))
def test_get_detail_unauthorized(self):
self.assertHttpForbidden(self.client.get(self.detail_uri))
self.assertHttpNotAuthorized(self.client.get(self.detail_uri))
def test_get_detail(self):
user = self.users[1]
@@ -342,12 +342,12 @@ class UserPreferenceViewSetTest(CacheIsolationTestCase, UserApiTestCase):
self.assertHttpMethodNotAllowed(self.request_with_auth("delete", self.LIST_URI))
def test_list_unauthorized(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
@override_settings(DEBUG=True)
@override_settings(EDX_API_KEY=None)
def test_debug_auth(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
def test_get_list_nonempty(self):
result = self.get_json(self.LIST_URI)
@@ -433,7 +433,7 @@ class UserPreferenceViewSetTest(CacheIsolationTestCase, UserApiTestCase):
self.assertHttpMethodNotAllowed(self.request_with_auth("delete", self.detail_uri))
def test_detail_unauthorized(self):
self.assertHttpForbidden(self.client.get(self.detail_uri))
self.assertHttpNotAuthorized(self.client.get(self.detail_uri))
def test_get_detail(self):
pref = self.prefs[1]
@@ -466,12 +466,12 @@ class PreferenceUsersListViewTest(UserApiTestCase):
self.assertHttpMethodNotAllowed(self.request_with_auth("delete", self.LIST_URI))
def test_unauthorized(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
@override_settings(DEBUG=True)
@override_settings(EDX_API_KEY=None)
def test_debug_auth(self):
self.assertHttpForbidden(self.client.get(self.LIST_URI))
self.assertHttpNotAuthorized(self.client.get(self.LIST_URI))
def test_get_basic(self):
result = self.get_json(self.LIST_URI)
@@ -583,8 +583,8 @@ class UpdateEmailOptInTestCase(UserAPITestCase, SharedModuleStoreTestCase):
def test_update_email_opt_in_anonymous_user(self):
"""
Test that an anonymous user gets 403 response when
updating email optin preference.
Test that an anonymous user gets 401 response when
updating email opt-in preference.
"""
self.client.logout()
response = self.client.post(self.url, {

30
openedx/core/djangoapps/util/management/commands/update_fixtures.py
View File

@@ -1,30 +0,0 @@
"""
Django management command to update the loaded test fixtures as necessary for
the current test environment. Currently just sets an appropriate domain for
each Site fixture.
"""
import os
from django.contrib.sites.models import Site
from django.core.management.base import BaseCommand
class Command(BaseCommand):
"""
update_fixtures management command
"""
help = "Update fixtures to match the current test environment."
def handle(self, *args, **options):
if 'BOK_CHOY_HOSTNAME' in os.environ:
# Fix the Site fixture domains so third party auth tests work correctly
host = os.environ['BOK_CHOY_HOSTNAME']
cms_port = os.environ['BOK_CHOY_CMS_PORT']
lms_port = os.environ['BOK_CHOY_LMS_PORT']
cms_domain = f'{host}:{cms_port}'
Site.objects.filter(name='cms').update(domain=cms_domain)
lms_domain = f'{host}:{lms_port}'
Site.objects.filter(name='lms').update(domain=lms_domain)

42
openedx/core/djangoapps/util/tests/test_update_fixtures.py
View File

@@ -1,42 +0,0 @@
""" # lint-amnesty, pylint: disable=django-not-configured
Tests of the update_fixtures management command for bok-choy test database
initialization.
"""
import os
import pytest
from django.contrib.sites.models import Site
from django.core.management import call_command
@pytest.fixture(scope='function')
def sites(db): # lint-amnesty, pylint: disable=unused-argument
Site.objects.create(name='cms', domain='localhost:8031')
Site.objects.create(name='lms', domain='localhost:8003')
def test_localhost(db, monkeypatch, sites): # lint-amnesty, pylint: disable=redefined-outer-name, unused-argument
monkeypatch.delitem(os.environ, 'BOK_CHOY_HOSTNAME', raising=False)
call_command('update_fixtures')
assert Site.objects.get(name='cms').domain == 'localhost:8031'
assert Site.objects.get(name='lms').domain == 'localhost:8003'
def test_devstack_cms(db, monkeypatch, sites): # lint-amnesty, pylint: disable=redefined-outer-name, unused-argument
monkeypatch.setitem(os.environ, 'BOK_CHOY_HOSTNAME', 'edx.devstack.cms')
monkeypatch.setitem(os.environ, 'BOK_CHOY_CMS_PORT', '18031')
monkeypatch.setitem(os.environ, 'BOK_CHOY_LMS_PORT', '18003')
call_command('update_fixtures')
assert Site.objects.get(name='cms').domain == 'edx.devstack.cms:18031'
assert Site.objects.get(name='lms').domain == 'edx.devstack.cms:18003'
def test_devstack_lms(db, monkeypatch, sites): # lint-amnesty, pylint: disable=redefined-outer-name, unused-argument
monkeypatch.setitem(os.environ, 'BOK_CHOY_HOSTNAME', 'edx.devstack.lms')
monkeypatch.setitem(os.environ, 'BOK_CHOY_CMS_PORT', '18031')
monkeypatch.setitem(os.environ, 'BOK_CHOY_LMS_PORT', '18003')
call_command('update_fixtures')
assert Site.objects.get(name='cms').domain == 'edx.devstack.lms:18031'
assert Site.objects.get(name='lms').domain == 'edx.devstack.lms:18003'

22
openedx/core/djangoapps/video_config/toggles.py
View File

@@ -15,3 +15,25 @@ WAFFLE_FLAG_NAMESPACE = 'video_config'
PUBLIC_VIDEO_SHARE = CourseWaffleFlag(
f'{WAFFLE_FLAG_NAMESPACE}.public_video_share', __name__
)
# .. toggle_name: video_config.transcript_feedback
# .. toggle_implementation: CourseWaffleFlag
# .. toggle_default: False
# .. toggle_description: Gates access to the transcript feedback widget feature.
# .. toggle_use_cases: temporary, opt_in
# .. toggle_creation_date: 2023-05-10
# .. toggle_target_removal_date: None
TRANSCRIPT_FEEDBACK = CourseWaffleFlag(
f'{WAFFLE_FLAG_NAMESPACE}.transcript_feedback', __name__
)
# .. toggle_name: video_config.xpert_translations_ui
# .. toggle_implementation: CourseWaffleFlag
# .. toggle_default: False
# .. toggle_description: Gates access to the Xpert Translations UI feature.
# .. toggle_use_cases: temporary, opt_in
# .. toggle_creation_date: 2023-10-11
# .. toggle_target_removal_date: None
XPERT_TRANSLATIONS_UI = CourseWaffleFlag(
f'{WAFFLE_FLAG_NAMESPACE}.xpert_translations_ui', __name__
)

4
openedx/core/lib/api/test_utils.py
View File

@@ -64,6 +64,10 @@ class ApiTestCase(TestCase):
"""Assert that the given response has the status code 201"""
assert response.status_code == 201
def assertHttpNotAuthorized(self, response):
"""Assert that the given response has the status code 401"""
assert response.status_code == 401
def assertHttpForbidden(self, response):
"""Assert that the given response has the status code 403"""
assert response.status_code == 403

28
openedx/core/lib/events.py
View File

@@ -1,28 +0,0 @@
"""Temporary method for use in rolling out a new event producer configuration."""
from django.conf import settings
from edx_django_utils.monitoring import set_custom_attribute
def determine_producer_config_for_signal_and_topic(signal, topic):
"""
Utility method to determine the setting for the given signal and topic in EVENT_BUS_PRODUCER_CONFIG
Records to New Relic for later analysis.
Parameters
signal (OpenEdxPublicSignal): The signal being sent to the event bus
topic (string): The topic to which the signal is being sent (without environment prefix)
Returns
True if the signal is enabled for that topic in EVENT_BUS_PRODUCER_CONFIG
False if the signal is explicitly disabled for that topic in EVENT_BUS_PRODUCER_CONFIG
None if the signal/topic pair is not present in EVENT_BUS_PRODUCER_CONFIG
"""
event_type_producer_configs = getattr(settings, "EVENT_BUS_PRODUCER_CONFIG",
{}).get(signal.event_type, {})
topic_config = event_type_producer_configs.get(topic, {})
topic_setting = topic_config.get('enabled', None)
set_custom_attribute(f'producer_config_setting_{topic}_{signal.event_type}',
topic_setting if topic_setting is not None else 'Unset')
return topic_setting

32
openedx/core/lib/logsettings.py
View File

@@ -126,6 +126,38 @@ def log_python_warnings():
warnings.filterwarnings('ignore', 'Setting _field_data is deprecated')
warnings.filterwarnings('ignore', 'Setting _field_data via the constructor is deprecated')
warnings.filterwarnings('ignore', '.*unclosed.*', category=ResourceWarning)
# Remove default_app_config warning after updating Django to 4.2
warnings.filterwarnings(
'ignore',
'.*You can remove default_app_config.*',
category=PendingDeprecationWarning
)
warnings.filterwarnings(
'ignore',
'Instead access HTTPResponse.headers directly.*',
category=DeprecationWarning,
module='elasticsearch'
)
warnings.filterwarnings(
'ignore',
'Using or importing the ABCs from \'collections\' instead of from \'collections.abc\' is deprecated.*',
category=DeprecationWarning,
module="sass",
)
warnings.filterwarnings(
'ignore',
'Deprecated call to `pkg_resources.declare_namespace.*',
category=DeprecationWarning,
)
warnings.filterwarnings(
'ignore',
'.*pkg_resources is deprecated as an API.*',
category=DeprecationWarning,
)
warnings.filterwarnings(
'ignore', "'etree' is deprecated. Use 'xml.etree.ElementTree' instead.",
category=DeprecationWarning, module='wiki'
)
# try:
# # There are far too many of these deprecation warnings in startup to output for every management command;
# # suppress them until we've fixed at least the most common ones as reported by the test suite