diff --git a/server/apps/research/admin/article_admin.py b/server/apps/research/admin/article_admin.py index 352ae59..689ec34 100644 --- a/server/apps/research/admin/article_admin.py +++ b/server/apps/research/admin/article_admin.py @@ -1,11 +1,40 @@ from django.contrib import admin from django import forms -from apps.research.models import Article, ArticleSlugHistory +from apps.research.models import Article, ArticleSlugHistory, RelatedArticle from tinymce.widgets import TinyMCE from .slug_history import current_slug_history + +class RelatedArticleInline(admin.TabularInline): + model = RelatedArticle + fk_name = 'from_article' + extra = 1 + max_num = 3 + verbose_name = 'Related Article' + verbose_name_plural = 'Related Articles' + def formfield_for_foreignkey(self, db_field, request, **kwargs): + if db_field.name == 'to_article': + # Get the parent object (Article) from the request + obj_id = request.resolver_match.kwargs.get('object_id') + # For new articles (when obj_id is None), show all ready articles + base_queryset = Article.objects.filter(status='ready') + + if obj_id: + try: + parent_obj = Article.objects.get(pk=obj_id) + # Exclude self-reference and articles that already have a relationship + base_queryset = base_queryset.exclude( + id=parent_obj.id + ).exclude( + related_from__to_article=parent_obj + ) + except Article.DoesNotExist: + pass + + kwargs['queryset'] = base_queryset + return super().formfield_for_foreignkey(db_field, request, **kwargs) class ArticleForm(forms.ModelForm): class Meta: model = Article @@ -19,10 +48,26 @@ def __init__(self, *args, **kwargs): class ArticleAdmin(admin.ModelAdmin): """Admin interface for the Article model.""" form = ArticleForm + inlines = [RelatedArticleInline] + def current_slug_history(self, obj): return current_slug_history(obj) current_slug_history.short_description = 'Slug Change History' + def get_inlines(self, request, obj): + # Allow inlines for both new and existing articles + return [RelatedArticleInline] + + def save_related(self, request, form, formsets, change): + """Handle saving related articles for both new and existing articles.""" + super().save_related(request, form, formsets, change) + + # Process related articles from inline formsets + for formset in formsets: + if isinstance(formset, RelatedArticleInline): + # The related articles will be saved automatically through the formset + pass + fieldsets = [ ('Article Details', {'fields': ['title', 'slug', 'authors', 'acknowledgement', 'categories', 'thumb', 'content', 'summary', 'status', 'scheduled_publish_time']}), ('Sponsorship Details', {'fields': ['is_sponsored', 'sponsor_color', 'sponsor_text_color']}), diff --git a/server/apps/research/migrations/0015_relatedarticle_article_related_articles_and_more.py b/server/apps/research/migrations/0015_relatedarticle_article_related_articles_and_more.py new file mode 100644 index 0000000..e8dc4b9 --- /dev/null +++ b/server/apps/research/migrations/0015_relatedarticle_article_related_articles_and_more.py @@ -0,0 +1,52 @@ +# Generated by Django 5.0.8 on 2024-12-12 04:31 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('research', '0014_alter_article_authors'), + ] + + operations = [ + migrations.CreateModel( + name='RelatedArticle', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('from_article', models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name='related_from', + to='research.article', + )), + ('to_article', models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name='related_to', + to='research.article', + )), + ], + ), + migrations.AddField( + model_name='article', + name='related_articles', + field=models.ManyToManyField( + blank=True, + related_name='referenced_by', + through='research.RelatedArticle', + to='research.article', + ), + ), + migrations.AddConstraint( + model_name='relatedarticle', + constraint=models.CheckConstraint( + check=~models.Q(from_article=models.F('to_article')), + name='prevent_self_reference', + ), + ), + migrations.AlterUniqueTogether( + name='relatedarticle', + unique_together={('from_article', 'to_article')}, + ), + ] diff --git a/server/apps/research/models/__init__.py b/server/apps/research/models/__init__.py index 3737c10..2cf7bc8 100644 --- a/server/apps/research/models/__init__.py +++ b/server/apps/research/models/__init__.py @@ -1,3 +1,3 @@ from .category import Category from .author import Author -from .article import Article, ArticleSlugHistory \ No newline at end of file +from .article import Article, ArticleSlugHistory, RelatedArticle \ No newline at end of file diff --git a/server/apps/research/models/article.py b/server/apps/research/models/article.py index 72cca9d..88c3108 100644 --- a/server/apps/research/models/article.py +++ b/server/apps/research/models/article.py @@ -1,5 +1,6 @@ from django.db import models from django.utils.text import slugify +from django.core.exceptions import ValidationError from apps.common.models import BaseModel from apps.research.managers import ArticleObjects from .category import Category @@ -12,28 +13,41 @@ import uuid from django.db import transaction + def get_default_thumb(): return f"{settings.MEDIA_URL}images/2077-Collective.png" + class Article(BaseModel): """Model for articles.""" - + options = ( - ('draft', 'Draft'), - ('ready', 'Ready'), + ("draft", "Draft"), + ("ready", "Ready"), ) title = models.TextField() content = HTMLField(blank=True, null=True) summary = models.TextField(blank=True) acknowledgement = HTMLField(blank=True, null=True) - authors = models.ManyToManyField(Author, blank=True, related_name='articles') + authors = models.ManyToManyField(Author, blank=True, related_name="articles") slug = models.SlugField(max_length=255, blank=True, db_index=True) - categories = models.ManyToManyField(Category, blank=True, related_name='articles') - thumb = models.ImageField(upload_to='images/', default=get_default_thumb, blank=True) + categories = models.ManyToManyField(Category, blank=True, related_name="articles") + related_articles = models.ManyToManyField( + "self", + blank=True, + symmetrical=False, + related_name="referenced_by", + through="RelatedArticle", + ) + thumb = models.ImageField( + upload_to="images/", default=get_default_thumb, blank=True + ) views = models.PositiveBigIntegerField(default=0) - status = models.CharField(max_length=10, choices=options, default='draft', db_index=True) - scheduled_publish_time = models.DateTimeField(null=True, blank=True, db_index=True) + status = models.CharField( + max_length=10, choices=options, default="draft", db_index=True + ) + scheduled_publish_time = models.DateTimeField(null=True, blank=True, db_index=True) table_of_contents = models.JSONField(default=list, blank=True) is_sponsored = models.BooleanField(default=False) sponsor_color = models.CharField(max_length=7, default="#FF0420") @@ -43,8 +57,31 @@ class Article(BaseModel): post_objects = ArticleObjects() class Meta: - ordering = ('-scheduled_publish_time',) - + ordering = ("-scheduled_publish_time",) + + def get_related_articles(self): + """ + Get manually selected related articles for this article. + If none exist, fallback to default frontend logic + """ + + manual_related_articles = self.related_articles.filter(status="ready").order_by( + "-scheduled_publish_time" + )[:3] + + if manual_related_articles.exists(): + return manual_related_articles + + # Fallback to last 3 articles in the same category + + category_articles = ( + Article.objects.filter(categories__in=self.categories.all(), status="ready") + .exclude(id=self.id) + .order_by("-scheduled_publish_time") + .distinct()[:3] + ) + return category_articles + def calculate_min_read(self): word_count = len(self.content.split()) words_per_minute = 300 # Average reading speed (words per minute) @@ -58,30 +95,38 @@ def __str__(self): def build_table_of_contents(self): """Build the table of contents from the article content.""" - soup = BeautifulSoup(self.content, 'html.parser') - headers = soup.find_all(['h1', 'h2', 'h3']) - + soup = BeautifulSoup(self.content, "html.parser") + headers = soup.find_all(["h1", "h2", "h3"]) + toc = [] - stack = [{'level': 0, 'children': toc}] + stack = [{"level": 0, "children": toc}] for header in headers: level = int(header.name[1]) title = header.get_text() - header['id'] = slugify(title) + header["id"] = slugify(title) - while level <= stack[-1]['level']: + while level <= stack[-1]["level"]: stack.pop() - new_item = {'title': title, 'id': header['id'], 'children': []} - - stack[-1]['children'].append(new_item) - stack.append({'level': level, 'children': new_item['children']}) + new_item = {"title": title, "id": header["id"], "children": []} + + stack[-1]["children"].append(new_item) + stack.append({"level": level, "children": new_item["children"]}) self.table_of_contents = toc self.content = str(soup) def save(self, *args, **kwargs): """Override the save method to generate a unique slug and build table of contents.""" + is_new = self.pk is None + temp_related_articles = [] + + # If this is a new article and there are related articles in the form + if is_new and hasattr(self, "_temp_related_articles"): + temp_related_articles = self._temp_related_articles + delattr(self, "_temp_related_articles") + if not self.slug or self.title_update(): self.slug = self.generate_unique_slug() @@ -92,25 +137,39 @@ def save(self, *args, **kwargs): # Generate new slug first if not self.slug or self.title_update(): self.slug = self.generate_unique_slug() - + # Then check if we need to create slug history if old_instance.slug and old_instance.slug != self.slug: # Only create history if the slug actually changed and isn't empty with transaction.atomic(): ArticleSlugHistory.objects.create( - article=self, - old_slug=old_instance.slug + article=self, old_slug=old_instance.slug ) except Article.DoesNotExist: - pass - + pass + if self.content: self.build_table_of_contents() - - if self.scheduled_publish_time and self.status == 'draft' and timezone.now() >= self.scheduled_publish_time: - self.status = 'ready' - super().save(*args, **kwargs) + + with transaction.atomic(): + super().save(*args, **kwargs) + + # If this was a new article and we had temporary related articles + if is_new and temp_related_articles: + for related_article in temp_related_articles: + RelatedArticle.objects.create( + from_article=self, to_article=related_article + ) + + def set_temp_related_articles(self, related_articles): + """ + Store related articles temporarily before the initial save. + + Args: + related_articles: List of Article instances to be related + """ + self._temp_related_articles = related_articles def generate_unique_slug(self): """Generate a unique slug for the article.""" @@ -122,28 +181,81 @@ def generate_unique_slug(self): slug = f"{base_slug}-{num}" num += 1 return slug - + def title_update(self): """Check if the title has changed.""" if self.pk: # Only check if the article exists - original = Article.objects.filter(pk=self.pk).only('title').first() + original = Article.objects.filter(pk=self.pk).only("title").first() if original: return original.title != self.title return False +class RelatedArticle(models.Model): + """Through model for related articles to prevent circular references.""" + + from_article = models.ForeignKey( + Article, on_delete=models.CASCADE, related_name="related_from" + ) + to_article = models.ForeignKey( + Article, on_delete=models.CASCADE, related_name="related_to" + ) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + unique_together = ("from_article", "to_article") + constraints = [ + models.CheckConstraint( + check=~models.Q(from_article=models.F("to_article")), + name="prevent_self_reference", + ), + ] + + def clean(self): + if not self.from_article.pk: + return + + # Prevent direct circular references + if RelatedArticle.objects.filter( + from_article=self.to_article, + to_article=self.from_article + ).exists(): + raise ValidationError("Circular references detected.") + + # Maximum of 3 related articles (only check for new records) + if not self.pk and RelatedArticle.objects.filter( + from_article=self.from_article + ).count() >= 3: + raise ValidationError("Maximum of 3 related articles allowed.") + + def save(self, *args, **kwargs): + with transaction.atomic(): + if self.from_article.pk: + # Lock all related records for this from_article + locked_relations = RelatedArticle.objects.select_for_update().filter( + from_article=self.from_article + ) + # Force evaluation of the queryset to acquire the lock + list(locked_relations) + + self.clean() + super().save(*args, **kwargs) + class ArticleSlugHistory(models.Model): """Model to track historical slugs for articles.""" + id = models.AutoField(primary_key=True) - article = models.ForeignKey('Article', on_delete=models.CASCADE, related_name='slug_history') + article = models.ForeignKey( + "Article", on_delete=models.CASCADE, related_name="slug_history" + ) old_slug = models.SlugField(max_length=255, db_index=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: - unique_together = ('article', 'old_slug') + unique_together = ("article", "old_slug") indexes = [ - models.Index(fields=['old_slug']), + models.Index(fields=["old_slug"]), ] - db_table = 'research_articleslughistory' # explicitly set table name + db_table = "research_articleslughistory" # explicitly set table name def __str__(self): return f"{self.old_slug} -> {self.article.slug}" \ No newline at end of file diff --git a/server/apps/research/serializers/article_serializer.py b/server/apps/research/serializers/article_serializer.py index 2bf8989..357556b 100644 --- a/server/apps/research/serializers/article_serializer.py +++ b/server/apps/research/serializers/article_serializer.py @@ -1,82 +1,219 @@ from rest_framework import serializers -from ..models import Article, Author, Category +from django.db import transaction +import logging +from django.conf import settings +from django.core.exceptions import ValidationError as DjangoValidationError +from ..models import Article, Author, Category, RelatedArticle from .author_serializer import AuthorSerializer from .category_serializer import CategorySerializer +logger = logging.getLogger(__name__) + class ArticleListSerializer(serializers.ModelSerializer): categories = CategorySerializer(many=True) authors = AuthorSerializer(many=True) class Meta: model = Article - include = ['categories' 'authors'] - exclude = [ - 'content', 'scheduled_publish_time', 'acknowledgement', - 'status', 'views', 'created_at', 'updated_at', 'table_of_contents' - ] + fields = ['id', 'title', 'slug', 'categories', 'authors'] -class ArticleSerializer(serializers.ModelSerializer): - """Serializer for the Article model.""" - authors = AuthorSerializer(many=True, read_only=True) - categories = CategorySerializer(many=True) - views = serializers.ReadOnlyField() - min_read = serializers.ReadOnlyField() +class RelatedArticleSerializer(serializers.ModelSerializer): + id = serializers.PrimaryKeyRelatedField(source='to_article.id', read_only=True) + title = serializers.CharField(source='to_article.title', read_only=True) + slug = serializers.CharField(source='to_article.slug', read_only=True) + thumb = serializers.ImageField(source='to_article.thumb', read_only=True) + class Meta: + model = RelatedArticle + fields = ['id', 'title', 'slug', 'thumb'] + +class ArticleSerializer(serializers.ModelSerializer): + def _handle_error(self, error, operation_type, context_data): + """ + Centralized error handling for article operations + """ + if isinstance(error, DjangoValidationError): + if settings.DEBUG: + raise serializers.ValidationError(error.message_dict) from error + raise serializers.ValidationError({ + "non_field_errors": ["Invalid data provided."] + }) from error + + logger.error( + f"Error {operation_type} article", + extra={**context_data, "error": str(error)}, + exc_info=True + ) + raise serializers.ValidationError({ + "non_field_errors": [f"Unable to {operation_type} article. Please try again later."] + }) from error + class Meta: model = Article fields = [ - 'id', 'slug', 'title', 'authors', 'thumb', 'categories', 'summary', - 'acknowledgement', 'content', 'min_read', 'status', 'views', - 'created_at', 'updated_at', 'scheduled_publish_time', 'table_of_contents', - 'is_sponsored', 'sponsor_color', 'sponsor_text_color' + 'id', 'title', 'slug', 'categories', 'authors', 'thumb', + 'content', 'summary', 'acknowledgement', 'status', + 'scheduled_publish_time', 'is_sponsored', 'sponsor_color', + 'sponsor_text_color', 'views', 'related_articles', 'table_of_contents' ] + def create(self, validated_data): + try: + return super().create(validated_data) + except Exception as e: + self._handle_error(e, "creating", {"validated_data": validated_data}) + + def update(self, instance, validated_data): + try: + return super().update(instance, validated_data) + except Exception as e: + self._handle_error(e, "updating", { + "instance_id": instance.id if instance else None, + "validated_data": validated_data + }) + class ArticleCreateUpdateSerializer(serializers.ModelSerializer): """Serializer for creating and updating articles.""" authors = serializers.PrimaryKeyRelatedField(queryset=Author.objects.all(), many=True, required=False) categories = serializers.PrimaryKeyRelatedField(queryset=Category.objects.all(), many=True, required=False) + related_article_ids = serializers.PrimaryKeyRelatedField( + queryset=Article.objects.filter(status='ready'), + many=True, + required=False, + write_only=True + ) class Meta: model = Article - fields = ['title', 'slug', 'categories', 'thumb', 'content', 'summary', 'acknowledgement', 'status', 'authors', 'scheduled_publish_time', 'is_sponsored', 'sponsor_color', 'sponsor_text_color'] + fields = [ + 'title', 'slug', 'categories', 'thumb', 'content', 'summary', + 'acknowledgement', 'status', 'authors', 'scheduled_publish_time', + 'is_sponsored', 'sponsor_color', 'sponsor_text_color', 'related_article_ids' + ] + + def _handle_error(self, error, operation_type, context_data): + """ + Centralized error handling for article operations + """ + if isinstance(error, DjangoValidationError): + raise serializers.ValidationError(error.message_dict) from error + + logger.error( + f"Error {operation_type} article", + extra={**context_data, "error": str(error)}, + exc_info=True + ) + raise serializers.ValidationError({ + "non_field_errors": [f"Unable to {operation_type} article. Please try again later."] + }) from error + + def validate_related_article_ids(self, value): + """Validate related articles.""" + if len(value) > 3: + raise serializers.ValidationError("You can only have a maximum of 3 related articles.") + + # Check for duplicates + article_ids = [article.id for article in value] + if len(article_ids) != len(set(article_ids)): + raise serializers.ValidationError("Duplicate articles are not allowed.") + + article_id = self.instance.id if self.instance else None + + if article_id and article_id in [article.id for article in value]: + raise serializers.ValidationError("An article cannot be related to itself.") - def create(self, validated_data: dict) -> Article: - """Create a new article instance.""" - request = self.context.get('request') + return value + + def _handle_relations(self, article, authors, categories, related_article_ids): + """ + Handle setting related objects for an article. + + Args: + article: The Article instance to update + authors: List of Author instances to associate + categories: List of Category instances to associate + related_article_ids: List of Article instances to set as related articles + + Raises: + ValidationError: If related objects can't be found or other errors occur + """ + try: + if authors: + article.authors.set(authors) + if categories: + article.categories.set(categories) + + if related_article_ids is not None: + with transaction.atomic(): + # Lock the article for update + article = Article.objects.select_for_update().get(id=article.id) + # Delete existing relations first + RelatedArticle.objects.filter(from_article=article).delete() + + # Create new relations + RelatedArticle.objects.bulk_create([ + RelatedArticle(from_article=article, to_article=related_article) + for related_article in related_article_ids + ]) + + except (Article.DoesNotExist, Author.DoesNotExist, Category.DoesNotExist) as e: + logger.error( + "Related object not found", + extra={ + "article_id": article.id, + "error": str(e) + }, + exc_info=True + ) + raise serializers.ValidationError("Related object not found") from e + + except Exception as e: + logger.error( + "Error handling article relations", + extra={ + "article_id": article.id, + "authors": [a.id for a in authors] if authors else None, + "categories": [c.id for c in categories] if categories else None, + "related_articles": [r.id for r in related_article_ids] if related_article_ids else None, + "error": str(e) + }, + exc_info=True + ) + raise serializers.ValidationError("An error occurred while setting related objects") from e + + def create(self, validated_data): + """Create a new Article instance.""" authors = validated_data.pop('authors', []) categories = validated_data.pop('categories', []) + related_article_ids = validated_data.pop('related_article_ids', []) try: + request = self.context.get('request') if not authors and request and hasattr(request, 'user'): user_author = Author.objects.filter(user=request.user).first() if user_author: authors = [user_author] - article = Article(**validated_data) - article.save() - - if authors: - article.authors.set(authors) - if categories: - article.categories.set(categories) - + article = Article.objects.create(**validated_data) + self._handle_relations(article, authors, categories, related_article_ids) return article - except Exception as e: - raise serializers.ValidationError(f"Error creating article: {str(e)}") + except Exception as e: + self._handle_error(e, "creating", {"validated_data": validated_data}) def update(self, instance: Article, validated_data: dict) -> Article: """Update an existing article instance.""" authors = validated_data.pop('authors', []) categories = validated_data.pop('categories', []) + related_article_ids = validated_data.pop('related_article_ids', None) + try: instance = super().update(instance, validated_data) - - if authors: - instance.authors.set(authors) - if categories: - instance.categories.set(categories) - + self._handle_relations(instance, authors, categories, related_article_ids) return instance - except Exception as e: - raise serializers.ValidationError(f"Error updating article: {str(e)}") + + except Exception as e: + self._handle_error(e, "updating", { + "instance_id": instance.id if instance else None, + "validated_data": validated_data + }) \ No newline at end of file diff --git a/server/apps/research/tasks.py b/server/apps/research/tasks.py index 34d25b0..9318d8e 100644 --- a/server/apps/research/tasks.py +++ b/server/apps/research/tasks.py @@ -1,11 +1,30 @@ from celery import shared_task from django.utils import timezone +from django.db import transaction from .models import Article +import logging + +logger = logging.getLogger(__name__) # TODO: Implement Querying the Articles in chunks in case of very large dataset -@shared_task +@shared_task(name='apps.research.tasks.publish_scheduled_articles') def publish_scheduled_articles(): - """Publish articles that are scheduled to be published.""" - now = timezone.now() - articles_to_publish = Article.objects.filter(status='draft', scheduled_publish_time__lte=now) - articles_to_publish.update(status='ready') \ No newline at end of file + """ + Update status of draft articles that have reached their scheduled publish time + """ + try: + with transaction.atomic(): + updated = Article.objects.filter( + status='draft', + scheduled_publish_time__isnull=False, + scheduled_publish_time__lte=timezone.now() + ).update(status='ready') + + if updated: + logger.info(f"Updated {updated} articles to ready status") + + return updated + + except Exception as e: + logger.error(f"Error publishing scheduled articles: {e}") + raise \ No newline at end of file diff --git a/server/apps/research/tests.py b/server/apps/research/tests.py index fb27fc1..165edf6 100644 --- a/server/apps/research/tests.py +++ b/server/apps/research/tests.py @@ -1,6 +1,10 @@ from django.test import TestCase from django.contrib.auth.models import User from django.utils.text import slugify +from django.core.exceptions import ValidationError +from django.db.utils import IntegrityError +from django.db import transaction +from .models import Article, RelatedArticle from .models import Article from datetime import datetime, timedelta from django.utils import timezone @@ -98,3 +102,114 @@ def test_increment_views(self): article.save() article.refresh_from_db() self.assertEqual(article.views, 1) + +class RelatedArticleModelTest(TestCase): + def setUp(self): + # Create test articles without author field + self.article1 = Article.objects.create( + title='Test Article 1', + content='Content 1', + status='ready' + ) + self.article2 = Article.objects.create( + title='Test Article 2', + content='Content 2', + status='ready' + ) + self.article3 = Article.objects.create( + title='Test Article 3', + content='Content 3', + status='ready' + ) + self.article4 = Article.objects.create( + title='Test Article 4', + content='Content 4', + status='ready' + ) + + def test_prevent_self_reference(self): + """Test that an article cannot be related to itself""" + with self.assertRaises(IntegrityError): + with transaction.atomic(): + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article1 + ) + + def test_prevent_circular_reference(self): + """Test that circular references are prevented""" + # Create first relationship + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article2 + ) + + # Attempt to create circular reference + with self.assertRaises(ValidationError): + RelatedArticle.objects.create( + from_article=self.article2, + to_article=self.article1 + ) + + def test_maximum_related_articles(self): + """Test that maximum of 3 related articles is enforced""" + # Create three related articles + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article2 + ) + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article3 + ) + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article4 + ) + + # Create fifth article and attempt to add it as fourth relation + article5 = Article.objects.create( + title='Test Article 5', + content='Content 5', + status='ready' + ) + + # Attempt to add fourth related article + with self.assertRaises(ValidationError): + RelatedArticle.objects.create( + from_article=self.article1, + to_article=article5 + ) + + def test_unique_relationships(self): + """Test that duplicate relationships are prevented""" + # Create first relationship + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article2 + ) + + # Attempt to create duplicate relationship + with self.assertRaises(IntegrityError): + with transaction.atomic(): + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article2 + ) + + def test_get_related_articles(self): + """Test the get_related_articles method""" + # Create related articles + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article2 + ) + RelatedArticle.objects.create( + from_article=self.article1, + to_article=self.article3 + ) + + related = self.article1.get_related_articles() + self.assertEqual(related.count(), 2) + self.assertIn(self.article2, related) + self.assertIn(self.article3, related) \ No newline at end of file