coulomb/vergabe-teilnahme
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

issue-facade integration

Browse source
This commit is contained in:
tegwick 2026-05-14 11:30:30 +02:00
parent c2b7cc64d6
commit a907ed6f74
16 changed files with 642 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

7
vergabe_teilnahme/apps/aufgaben/admin.py
View file

@ -24,8 +24,11 @@ class AufgabenVerknuepfungAdmin(admin.ModelAdmin):
@admin.register(ExternalIssue)
class ExternalIssueAdmin(admin.ModelAdmin):
list_display = ['aufgabe', 'system', 'issue_key', 'sync_status', 'letzter_sync']
list_filter = ['system', 'sync_status']
list_display = ['aufgabe', 'issue_facade_backend', 'issue_key',
'sync_status', 'letzter_sync']
list_filter = ['issue_facade_backend', 'sync_status']
readonly_fields = ['issue_facade_id', 'issue_key', 'issue_url',
'sync_status', 'letzter_sync', 'erstellt_am']
@admin.register(Bieterfrage)

10
vergabe_teilnahme/apps/aufgaben/forms.py
View file

@ -83,11 +83,9 @@ class AufgabenVerknuepfungForm(forms.Form):
class ExternalIssueForm(forms.ModelForm):
class Meta:
model = ExternalIssue
fields = ['system', 'issue_url', 'issue_key', 'notizen']
model = ExternalIssue
fields = ['notizen']
widgets = {
'system': forms.Select(attrs={'class': 'form-input'}),
'issue_url': forms.URLInput(attrs={'class': 'form-input', 'placeholder': 'https://...'}),
'issue_key': forms.TextInput(attrs={'class': 'form-input', 'placeholder': 'z.B. GH-42'}),
'notizen': forms.Textarea(attrs={'class': 'form-input', 'rows': 2}),
'notizen': forms.Textarea(attrs={'class': 'form-input', 'rows': 2,
'placeholder': 'Notizen (optional)'}),
}

43
vergabe_teilnahme/apps/aufgaben/issue_backends.py Normal file
View file

@ -0,0 +1,43 @@
from contextlib import contextmanager
from pathlib import Path
@contextmanager
def local_backend():
from django.conf import settings
from issue_tracker.backends.local import LocalSQLiteBackend
db_path = str(getattr(settings, 'ISSUE_FACADE_LOCAL_DB', '.issue-facade/issues.db'))
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
b = LocalSQLiteBackend()
b.connect({'db_path': db_path})
try:
yield b
finally:
b.disconnect()
@contextmanager
def remote_backend():
"""Yields GiteaBackend wenn konfiguriert, sonst None."""
from django.conf import settings
from issue_tracker.backends.gitea import GiteaBackend
cfg = getattr(settings, 'ISSUE_FACADE_GITEA', None)
if not cfg:
yield None
return
b = GiteaBackend()
b.connect(cfg)
try:
yield b
finally:
b.disconnect()
def gitea_configured() -> bool:
from django.conf import settings
cfg = getattr(settings, 'ISSUE_FACADE_GITEA', None)
return bool(cfg and cfg.get('token'))

127
vergabe_teilnahme/apps/aufgaben/issue_facade.py
View file

@ -1,28 +1,111 @@
from abc import ABC, abstractmethod
from datetime import datetime, timezone
from issue_tracker.core.models import Issue, IssueState, Label, Priority
from .issue_backends import gitea_configured, local_backend, remote_backend
class IssueAdapter(ABC):
def aufgabe_zu_issue(aufgabe) -> Issue:
"""Konvertiert eine Aufgabe in ein issue-facade Issue-Objekt."""
prioritaet_map = {1: Priority.HIGH, 2: Priority.MEDIUM, 3: Priority.LOW}
labels = [
Label(name='task'),
Label(name=f'priority:{prioritaet_map.get(aufgabe.prioritaet, Priority.MEDIUM).value}'),
]
if aufgabe.typ:
labels.append(Label(name=aufgabe.typ))
now = datetime.now(timezone.utc)
return Issue(
id='',
number=0,
title=aufgabe.titel,
description=aufgabe.beschreibung or '',
state=IssueState.OPEN,
created_at=now,
updated_at=now,
labels=labels,
)
def lokales_issue_erstellen(aufgabe) -> dict:
"""
Adapter-Basisklasse. Jeder externe Issue-Tracker implementiert diese
Schnittstelle. Registrierung via ISSUE_ADAPTERS-Dict in settings.
Legt ein Issue im lokalen SQLite-Backend an.
Gibt {'issue_facade_id', 'issue_key', 'sync_status'} zurück.
"""
@abstractmethod
def create_issue(self, aufgabe) -> dict:
"""Legt ein Issue im externen System an. Gibt {'url', 'key'} zurück."""
@abstractmethod
def fetch_status(self, external_issue) -> str:
"""Liest den aktuellen Status aus dem externen System."""
@abstractmethod
def close_issue(self, external_issue) -> None:
"""Schließt das Issue im externen System."""
issue = aufgabe_zu_issue(aufgabe)
with local_backend() as b:
created = b.create_issue(issue)
return {
'issue_facade_id': str(created.id),
'issue_key': f'#{created.number}',
'sync_status': created.state.value,
}
def get_adapter(system: str) -> 'IssueAdapter | None':
"""Gibt den registrierten Adapter für `system` zurück, oder None."""
from django.conf import settings
adapters = getattr(settings, 'ISSUE_ADAPTERS', {})
cls = adapters.get(system)
return cls() if cls else None
def an_gitea_delegieren(aufgabe, external_issue) -> dict:
"""
Schiebt ein lokales Issue nach Gitea.
Gibt {'issue_facade_backend', 'issue_facade_id', 'issue_url', 'issue_key', 'sync_status'} zurück.
Wirft ValueError wenn Gitea nicht konfiguriert ist.
"""
with remote_backend() as b:
if b is None:
raise ValueError('Gitea nicht konfiguriert (ISSUE_FACADE_GITEA fehlt in settings)')
issue = aufgabe_zu_issue(aufgabe)
created = b.create_issue(issue)
url = ''
if created.sync_metadata:
url = created.sync_metadata.get('url', '')
return {
'issue_facade_backend': 'gitea',
'issue_facade_id': str(created.number),
'issue_url': url,
'issue_key': f'#{created.number}',
'sync_status': created.state.value,
}
def status_synchronisieren(external_issue) -> str:
"""
Liest den aktuellen Status aus dem konfigurierten Backend.
Gibt den neuen sync_status-String zurück.
"""
from django.utils import timezone as dj_tz
backend_ctx = (
remote_backend() if external_issue.issue_facade_backend == 'gitea'
else local_backend()
)
with backend_ctx as b:
if b is None:
raise ValueError('Backend nicht verfügbar')
issue = b.get_issue(external_issue.issue_facade_id)
neuer_status = issue.state.value if issue else 'error'
external_issue.sync_status = neuer_status
external_issue.letzter_sync = dj_tz.now()
external_issue.save(update_fields=['sync_status', 'letzter_sync'])
return neuer_status
def issue_schliessen(external_issue) -> None:
"""Setzt das Issue im Backend auf CLOSED."""
backend_ctx = (
remote_backend() if external_issue.issue_facade_backend == 'gitea'
else local_backend()
)
with backend_ctx as b:
if b is None:
return
issue = b.get_issue(external_issue.issue_facade_id)
if issue:
issue.state = IssueState.CLOSED
b.update_issue(issue)
external_issue.sync_status = 'closed'
external_issue.save(update_fields=['sync_status'])
def get_adapter(system: str):
"""Rückwärtskompatible Stub — gibt None zurück (kein manuelles Adapter-Dict mehr)."""
return None

37
vergabe_teilnahme/apps/aufgaben/migrations/0006_external_issue_redesign.py Normal file
View file

@ -0,0 +1,37 @@
# Generated by Django 6.0.5 on 2026-05-14 09:17
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('aufgaben', '0005_verknuepfung_externalissue'),
]
operations = [
migrations.RemoveField(
model_name='externalissue',
name='system',
),
migrations.AddField(
model_name='externalissue',
name='issue_facade_backend',
field=models.CharField(choices=[('local', 'Lokal (SQLite)'), ('gitea', 'Gitea')], default='local', max_length=20),
),
migrations.AddField(
model_name='externalissue',
name='issue_facade_id',
field=models.CharField(blank=True, help_text='UUID (lokal) oder Issue-Number (Gitea)', max_length=200),
),
migrations.AlterField(
model_name='externalissue',
name='issue_key',
field=models.CharField(blank=True, help_text='z.B. "#42" oder "PROJ-1234"', max_length=100),
),
migrations.AlterField(
model_name='externalissue',
name='sync_status',
field=models.CharField(choices=[('open', 'Offen'), ('in_progress', 'In Bearbeitung'), ('blocked', 'Blockiert'), ('closed', 'Geschlossen'), ('error', 'Sync-Fehler')], default='open', max_length=20),
),
]

48
vergabe_teilnahme/apps/aufgaben/models.py
View file

@ -154,36 +154,42 @@ class AufgabenVerknuepfung(models.Model):
class ExternalIssue(models.Model):
SYSTEM_CHOICES = [
('github', 'GitHub Issues'),
('jira', 'Jira'),
('linear', 'Linear'),
('azure', 'Azure DevOps'),
('sonstiges', 'Sonstiges'),
BACKEND_CHOICES = [
('local', 'Lokal (SQLite)'),
('gitea', 'Gitea'),
]
SYNC_STATUS_CHOICES = [
('manuell', 'Manuell'),
('offen', 'Offen (extern)'),
('geschlossen', 'Geschlossen (extern)'),
('fehler', 'Sync-Fehler'),
('open', 'Offen'),
('in_progress', 'In Bearbeitung'),
('blocked', 'Blockiert'),
('closed', 'Geschlossen'),
('error', 'Sync-Fehler'),
]
aufgabe = models.OneToOneField(
Aufgabe, on_delete=models.CASCADE, related_name='external_issue'
)
system = models.CharField(max_length=20, choices=SYSTEM_CHOICES)
issue_url = models.URLField(blank=True)
issue_key = models.CharField(max_length=100, blank=True,
help_text='z.B. "GH-42" oder "PROJ-1234"')
sync_status = models.CharField(max_length=20, choices=SYNC_STATUS_CHOICES,
default='manuell')
issue_facade_backend = models.CharField(
max_length=20, choices=BACKEND_CHOICES, default='local'
)
issue_facade_id = models.CharField(
max_length=200, blank=True,
help_text='UUID (lokal) oder Issue-Number (Gitea)'
)
issue_url = models.URLField(blank=True)
issue_key = models.CharField(max_length=100, blank=True,
help_text='z.B. "#42" oder "PROJ-1234"')
sync_status = models.CharField(
max_length=20, choices=SYNC_STATUS_CHOICES, default='open'
)
letzter_sync = models.DateTimeField(null=True, blank=True)
notizen = models.TextField(blank=True)
erstellt_am = models.DateTimeField(auto_now_add=True)
notizen = models.TextField(blank=True)
erstellt_am = models.DateTimeField(auto_now_add=True)
class Meta:
verbose_name = 'External Issue'
verbose_name_plural = 'External Issues'
verbose_name = 'External Issue'
verbose_name_plural = 'External Issues'
def __str__(self):
return f'{self.get_system_display()} {self.issue_key or self.issue_url}'
ident = self.issue_key or (self.issue_facade_id[:8] if self.issue_facade_id else '?')
return f'{self.get_issue_facade_backend_display()} #{ident}'

104
vergabe_teilnahme/apps/aufgaben/tests.py
View file

@ -168,26 +168,28 @@ def test_aufgaben_verknuepfung_loeschen(client):
# ─── ExternalIssue ────────────────────────────────────────────────────────────
@pytest.fixture
def tmp_issue_db(tmp_path, settings):
settings.ISSUE_FACADE_LOCAL_DB = tmp_path / 'test_issues.db'
settings.ISSUE_FACADE_GITEA = None
return settings.ISSUE_FACADE_LOCAL_DB
@pytest.mark.django_db
def test_external_issue_erstellen(client):
def test_external_issue_erstellen(client, tmp_issue_db):
aufgabe = AufgabeFactory()
url = reverse('ausschreibungen:aufgaben:external_issue',
kwargs={'ausschreibung_id': aufgabe.ausschreibung_id, 'pk': aufgabe.pk})
response = client.post(url, {
'system': 'github',
'issue_url': 'https://github.com/org/repo/issues/42',
'issue_key': 'GH-42',
'notizen': '',
}, HTTP_HX_REQUEST='true')
response = client.post(url, {'notizen': ''}, HTTP_HX_REQUEST='true')
assert response.status_code == 200
assert ExternalIssue.objects.filter(aufgabe=aufgabe, system='github').exists()
assert ExternalIssue.objects.filter(aufgabe=aufgabe, issue_facade_backend='local').exists()
@pytest.mark.django_db
def test_external_issue_loeschen(client):
aufgabe = AufgabeFactory()
ei = ExternalIssue.objects.create(
aufgabe=aufgabe, system='jira', issue_key='PROJ-1'
aufgabe=aufgabe, issue_facade_backend='local', issue_key='#1'
)
url = reverse('ausschreibungen:aufgaben:external_issue_loeschen',
kwargs={'ausschreibung_id': aufgabe.ausschreibung_id, 'pk': aufgabe.pk})
@ -201,3 +203,87 @@ def test_issue_adapter_interface():
from .issue_facade import get_adapter
assert get_adapter('github') is None
assert get_adapter('nichtexistent') is None
# ─── Issue-Facade Integration ─────────────────────────────────────────────────
@pytest.mark.django_db
def test_aufgabe_zu_issue_mapping(tmp_issue_db):
from .issue_facade import aufgabe_zu_issue
aufgabe = AufgabeFactory(titel='Test-Aufgabe', beschreibung='Beschreibung', prioritaet=1)
issue = aufgabe_zu_issue(aufgabe)
assert issue.title == 'Test-Aufgabe'
assert issue.description == 'Beschreibung'
label_names = [l.name for l in issue.labels]
assert 'task' in label_names
assert 'priority:high' in label_names
@pytest.mark.django_db
def test_lokales_issue_erstellen(tmp_issue_db):
from .issue_facade import lokales_issue_erstellen
aufgabe = AufgabeFactory()
daten = lokales_issue_erstellen(aufgabe)
assert daten['issue_facade_id']
assert daten['issue_key'].startswith('#')
assert daten['sync_status'] == 'open'
@pytest.mark.django_db
def test_status_synchronisieren(tmp_issue_db):
from .issue_facade import lokales_issue_erstellen, status_synchronisieren
aufgabe = AufgabeFactory()
daten = lokales_issue_erstellen(aufgabe)
ei = ExternalIssue.objects.create(
aufgabe=aufgabe,
issue_facade_backend='local',
issue_facade_id=daten['issue_facade_id'],
issue_key=daten['issue_key'],
sync_status=daten['sync_status'],
)
neuer_status = status_synchronisieren(ei)
assert neuer_status == 'open'
ei.refresh_from_db()
assert ei.sync_status == 'open'
assert ei.letzter_sync is not None
@pytest.mark.django_db
def test_external_issue_bearbeiten_view_erstellt_issue(client, tmp_issue_db):
aufgabe = AufgabeFactory()
url = reverse('ausschreibungen:aufgaben:external_issue',
kwargs={'ausschreibung_id': aufgabe.ausschreibung_id, 'pk': aufgabe.pk})
response = client.post(url, {'notizen': 'Test-Notiz'}, HTTP_HX_REQUEST='true')
assert response.status_code == 200
ei = ExternalIssue.objects.get(aufgabe=aufgabe)
assert ei.issue_facade_backend == 'local'
assert ei.issue_facade_id
assert ei.notizen == 'Test-Notiz'
@pytest.mark.django_db
def test_external_issue_sync_view(client, tmp_issue_db):
from .issue_facade import lokales_issue_erstellen
aufgabe = AufgabeFactory()
daten = lokales_issue_erstellen(aufgabe)
ei = ExternalIssue.objects.create(
aufgabe=aufgabe,
issue_facade_backend='local',
issue_facade_id=daten['issue_facade_id'],
issue_key=daten['issue_key'],
sync_status=daten['sync_status'],
)
url = reverse('ausschreibungen:aufgaben:external_issue_sync',
kwargs={'ausschreibung_id': aufgabe.ausschreibung_id, 'pk': aufgabe.pk})
response = client.post(url, {}, HTTP_HX_REQUEST='true')
assert response.status_code == 200
ei.refresh_from_db()
assert ei.sync_status == 'open'
@pytest.mark.django_db
def test_get_adapter_stub():
from .issue_facade import get_adapter
assert get_adapter('github') is None
assert get_adapter('gitea') is None

2
vergabe_teilnahme/apps/aufgaben/urls.py
View file

@ -18,4 +18,6 @@ urlpatterns = [
path('<int:pk>/issue/', views.external_issue_bearbeiten, name='external_issue'),
path('<int:pk>/issue/panel/', views.external_issue_panel, name='external_issue_panel'),
path('<int:pk>/issue/loeschen/', views.external_issue_loeschen, name='external_issue_loeschen'),
path('<int:pk>/issue/push/', views.external_issue_push_remote, name='external_issue_push'),
path('<int:pk>/issue/sync/', views.external_issue_sync, name='external_issue_sync'),
]

59
vergabe_teilnahme/apps/aufgaben/views.py
View file

@ -7,6 +7,7 @@ from vergabe_teilnahme.apps.ausschreibungen.models import Ausschreibung
from django.http import HttpResponse
from .issue_backends import gitea_configured as _gitea_configured
from .models import Aufgabe, AufgabenVerknuepfung, Bieterfrage, ExternalIssue
AKTIVE_STATUS = [
@ -413,6 +414,7 @@ def verknuepfung_loeschen(request, ausschreibung_id, pk, vk_pk):
def external_issue_bearbeiten(request, ausschreibung_id, pk):
from .forms import ExternalIssueForm
from .issue_facade import lokales_issue_erstellen
ausschreibung = get_object_or_404(Ausschreibung, pk=ausschreibung_id)
aufgabe = get_object_or_404(Aufgabe, pk=pk, ausschreibung=ausschreibung)
@ -423,11 +425,18 @@ def external_issue_bearbeiten(request, ausschreibung_id, pk):
if form.is_valid():
ei = form.save(commit=False)
ei.aufgabe = aufgabe
if not issue:
daten = lokales_issue_erstellen(aufgabe)
ei.issue_facade_backend = 'local'
ei.issue_facade_id = daten['issue_facade_id']
ei.issue_key = daten['issue_key']
ei.sync_status = daten['sync_status']
ei.save()
if _is_htmx(request):
return render(request, 'aufgaben/partials/external_issue_panel.html', {
'aufgabe': aufgabe,
'ausschreibung': ausschreibung,
'gitea_configured': _gitea_configured(),
})
return redirect('ausschreibungen:aufgaben:detail',
ausschreibung_id=ausschreibung_id, pk=pk)
@ -443,12 +452,61 @@ def external_issue_bearbeiten(request, ausschreibung_id, pk):
return redirect('ausschreibungen:aufgaben:detail', ausschreibung_id=ausschreibung_id, pk=pk)
def external_issue_push_remote(request, ausschreibung_id, pk):
"""Schiebt ein lokales Issue nach Gitea."""
from .issue_facade import an_gitea_delegieren
ausschreibung = get_object_or_404(Ausschreibung, pk=ausschreibung_id)
aufgabe = get_object_or_404(Aufgabe, pk=pk, ausschreibung=ausschreibung)
ei = get_object_or_404(ExternalIssue, aufgabe=aufgabe)
push_error = None
if request.method == 'POST':
try:
daten = an_gitea_delegieren(aufgabe, ei)
for k, v in daten.items():
setattr(ei, k, v)
ei.save()
except ValueError as exc:
push_error = str(exc)
if _is_htmx(request):
return render(request, 'aufgaben/partials/external_issue_panel.html', {
'aufgabe': aufgabe,
'ausschreibung': ausschreibung,
'gitea_configured': _gitea_configured(),
'push_error': push_error,
})
return redirect('ausschreibungen:aufgaben:detail', ausschreibung_id=ausschreibung_id, pk=pk)
def external_issue_sync(request, ausschreibung_id, pk):
"""Aktualisiert sync_status aus dem Backend."""
from .issue_facade import status_synchronisieren
ausschreibung = get_object_or_404(Ausschreibung, pk=ausschreibung_id)
aufgabe = get_object_or_404(Aufgabe, pk=pk, ausschreibung=ausschreibung)
ei = get_object_or_404(ExternalIssue, aufgabe=aufgabe)
if request.method == 'POST':
try:
status_synchronisieren(ei)
except Exception:
ei.sync_status = 'error'
ei.save(update_fields=['sync_status'])
if _is_htmx(request):
return render(request, 'aufgaben/partials/external_issue_panel.html', {
'aufgabe': aufgabe,
'ausschreibung': ausschreibung,
'gitea_configured': _gitea_configured(),
})
return redirect('ausschreibungen:aufgaben:detail', ausschreibung_id=ausschreibung_id, pk=pk)
def external_issue_panel(request, ausschreibung_id, pk):
ausschreibung = get_object_or_404(Ausschreibung, pk=ausschreibung_id)
aufgabe = get_object_or_404(Aufgabe, pk=pk, ausschreibung=ausschreibung)
return render(request, 'aufgaben/partials/external_issue_panel.html', {
'aufgabe': aufgabe,
'ausschreibung': ausschreibung,
'gitea_configured': _gitea_configured(),
})
@ -461,6 +519,7 @@ def external_issue_loeschen(request, ausschreibung_id, pk):
return render(request, 'aufgaben/partials/external_issue_panel.html', {
'aufgabe': aufgabe,
'ausschreibung': ausschreibung,
'gitea_configured': _gitea_configured(),
})
return redirect('ausschreibungen:aufgaben:detail',
ausschreibung_id=ausschreibung_id, pk=pk)