SonarQube Analysis กับ Hexagonal Architecture

SonarQube Code Quality Analysis

SonarQube วิเคราะห์ Code Quality ตรวจจับ Bugs, Code Smells, Vulnerabilities รองรับ 30+ ภาษา ใช้ Quality Gates กำหนดมาตรฐาน ใช้ร่วมกับ CI/CD Pipeline

Hexagonal Architecture แยก Domain ออกจาก External Systems ใช้ Ports and Adapters Pattern ทำให้ Code สะอาด ทดสอบง่าย SonarQube ช่วยตรวจสอบว่า Code ยังคง Clean

SonarQube Setup และ Configuration

# === SonarQube Setup ===

# 1. Docker Compose
# docker-compose.yml
# version: '3.8'
# services:
#   sonarqube:
#     image: sonarqube:community
#     ports:
#       - "9000:9000"
#     environment:
#       SONAR_JDBC_URL: jdbc:postgresql://db:5432/sonar
#       SONAR_JDBC_USERNAME: sonar
#       SONAR_JDBC_PASSWORD: sonar
#     volumes:
#       - sonarqube_data:/opt/sonarqube/data
#       - sonarqube_extensions:/opt/sonarqube/extensions
#     depends_on:
#       - db
#
#   db:
#     image: postgres:16
#     environment:
#       POSTGRES_USER: sonar
#       POSTGRES_PASSWORD: sonar
#       POSTGRES_DB: sonar
#     volumes:
#       - postgresql_data:/var/lib/postgresql/data

# docker compose up -d
# เข้า http://localhost:9000 (admin/admin)

# 2. SonarScanner CLI
# npm install -g sonarqube-scanner
# หรือ
# brew install sonar-scanner

# 3. sonar-project.properties
cat > sonar-project.properties << 'EOF'
sonar.projectKey=my-hexagonal-app
sonar.projectName=My Hexagonal App
sonar.projectVersion=1.0

# Source
sonar.sources=src
sonar.tests=tests
sonar.sourceEncoding=UTF-8

# Language
sonar.language=py
sonar.python.version=3.11

# Coverage
sonar.python.coverage.reportPaths=coverage.xml
sonar.python.xunit.reportPath=test-results.xml

# Exclusions
sonar.exclusions=**/migrations/**,**/tests/**,**/__pycache__/**
sonar.test.exclusions=**/migrations/**

# Quality Gate
sonar.qualitygate.wait=true
EOF

# 4. รัน Analysis
# sonar-scanner \
#   -Dsonar.host.url=http://localhost:9000 \
#   -Dsonar.login=your-token

# 5. GitHub Actions Integration
# .github/workflows/sonar.yml
# name: SonarQube Analysis
# on: [push, pull_request]
# jobs:
#   sonar:
#     runs-on: ubuntu-latest
#     steps:
#       - uses: actions/checkout@v4
#         with:
#           fetch-depth: 0
#       - uses: actions/setup-python@v5
#         with:
#           python-version: '3.11'
#       - run: |
#           pip install -r requirements.txt
#           pytest --cov=src --cov-report=xml tests/
#       - uses: SonarSource/sonarqube-scan-action@master
#         env:
#           SONAR_TOKEN: }
#           SONAR_HOST_URL: }

echo "SonarQube configured"
echo "  URL: http://localhost:9000"
echo "  Scanner: sonar-scanner CLI"
echo "  CI/CD: GitHub Actions integration"

Hexagonal Architecture Implementation

# === Hexagonal Architecture ด้วย Python ===
# Project Structure:
# src/
#   domain/         # Business Logic (ไม่พึ่ง External)
#     models.py
#     services.py
#     ports.py      # Interfaces
#   adapters/       # External Implementations
#     repositories/
#       postgres_user_repo.py
#       inmemory_user_repo.py
#     notifications/
#       email_service.py
#       slack_service.py
#   api/            # Input Adapters (REST, GraphQL)
#     routes.py
#   config/
#     container.py  # Dependency Injection

# === domain/models.py ===
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from enum import Enum

class UserRole(Enum):
    ADMIN = "admin"
    MEMBER = "member"
    VIEWER = "viewer"

@dataclass
class User:
    id: Optional[str] = None
    email: str = ""
    name: str = ""
    role: UserRole = UserRole.MEMBER
    is_active: bool = True
    created_at: datetime = field(default_factory=datetime.now)

    def activate(self):
        self.is_active = True

    def deactivate(self):
        self.is_active = False

    def promote(self, role: UserRole):
        if role == UserRole.ADMIN and self.role != UserRole.MEMBER:
            raise ValueError("Only members can be promoted to admin")
        self.role = role

# === domain/ports.py (Interfaces) ===
from abc import ABC, abstractmethod
from typing import List, Optional

class UserRepository(ABC):
    """Port สำหรับ User Data Access"""

    @abstractmethod
    def save(self, user: User) -> User:
        pass

    @abstractmethod
    def find_by_id(self, user_id: str) -> Optional[User]:
        pass

    @abstractmethod
    def find_by_email(self, email: str) -> Optional[User]:
        pass

    @abstractmethod
    def find_all(self, limit: int = 100, offset: int = 0) -> List[User]:
        pass

    @abstractmethod
    def delete(self, user_id: str) -> bool:
        pass

class NotificationService(ABC):
    """Port สำหรับ Notifications"""

    @abstractmethod
    def send(self, to: str, subject: str, body: str) -> bool:
        pass

class EventPublisher(ABC):
    """Port สำหรับ Domain Events"""

    @abstractmethod
    def publish(self, event_type: str, data: dict) -> None:
        pass

# === domain/services.py (Business Logic) ===
class UserService:
    """Domain Service — Business Logic เท่านั้น"""

    def __init__(self, repo: UserRepository, notifier: NotificationService,
                 publisher: EventPublisher):
        self.repo = repo
        self.notifier = notifier
        self.publisher = publisher

    def register_user(self, email: str, name: str) -> User:
        existing = self.repo.find_by_email(email)
        if existing:
            raise ValueError(f"Email already registered: {email}")

        user = User(email=email, name=name)
        saved = self.repo.save(user)

        self.notifier.send(email, "Welcome!", f"สวัสดี {name}")
        self.publisher.publish("user.registered", {"user_id": saved.id})

        return saved

    def get_user(self, user_id: str) -> User:
        user = self.repo.find_by_id(user_id)
        if not user:
            raise ValueError(f"User not found: {user_id}")
        return user

    def deactivate_user(self, user_id: str) -> User:
        user = self.get_user(user_id)
        user.deactivate()
        self.repo.save(user)
        self.publisher.publish("user.deactivated", {"user_id": user_id})
        return user

print("Hexagonal Architecture:")
print("  Domain: Models, Services, Ports (no external deps)")
print("  Adapters: Repositories, Notifications, API")
print("  Ports: UserRepository, NotificationService, EventPublisher")

Adapters และ Tests

# === Adapters (Implementations) ===

# adapters/repositories/inmemory_user_repo.py
import uuid
from typing import List, Optional

class InMemoryUserRepository:
    """In-memory Implementation สำหรับ Testing"""

    def __init__(self):
        self.users = {}

    def save(self, user):
        if not user.id:
            user.id = str(uuid.uuid4())
        self.users[user.id] = user
        return user

    def find_by_id(self, user_id):
        return self.users.get(user_id)

    def find_by_email(self, email):
        for user in self.users.values():
            if user.email == email:
                return user
        return None

    def find_all(self, limit=100, offset=0):
        all_users = list(self.users.values())
        return all_users[offset:offset + limit]

    def delete(self, user_id):
        if user_id in self.users:
            del self.users[user_id]
            return True
        return False

# adapters/notifications/fake_notifier.py
class FakeNotificationService:
    """Fake Notification สำหรับ Testing"""

    def __init__(self):
        self.sent = []

    def send(self, to, subject, body):
        self.sent.append({"to": to, "subject": subject, "body": body})
        return True

# adapters/events/fake_publisher.py
class FakeEventPublisher:
    """Fake Event Publisher สำหรับ Testing"""

    def __init__(self):
        self.events = []

    def publish(self, event_type, data):
        self.events.append({"type": event_type, "data": data})

# === Tests ===
# tests/test_user_service.py

def test_register_user():
    """ทดสอบ Register User"""
    repo = InMemoryUserRepository()
    notifier = FakeNotificationService()
    publisher = FakeEventPublisher()
    service = UserService(repo, notifier, publisher)

    user = service.register_user("test@example.com", "Test User")

    assert user.id is not None
    assert user.email == "test@example.com"
    assert user.name == "Test User"
    assert user.is_active is True
    assert len(notifier.sent) == 1
    assert len(publisher.events) == 1
    assert publisher.events[0]["type"] == "user.registered"
    print("  PASS: test_register_user")

def test_register_duplicate_email():
    """ทดสอบ Register Email ซ้ำ"""
    repo = InMemoryUserRepository()
    notifier = FakeNotificationService()
    publisher = FakeEventPublisher()
    service = UserService(repo, notifier, publisher)

    service.register_user("test@example.com", "User 1")

    try:
        service.register_user("test@example.com", "User 2")
        assert False, "Should raise ValueError"
    except ValueError:
        pass
    print("  PASS: test_register_duplicate_email")

def test_deactivate_user():
    """ทดสอบ Deactivate User"""
    repo = InMemoryUserRepository()
    notifier = FakeNotificationService()
    publisher = FakeEventPublisher()
    service = UserService(repo, notifier, publisher)

    user = service.register_user("test@example.com", "Test")
    deactivated = service.deactivate_user(user.id)

    assert deactivated.is_active is False
    print("  PASS: test_deactivate_user")

# รัน Tests
print("\nRunning Tests:")
test_register_user()
test_register_duplicate_email()
test_deactivate_user()
print("\nAll tests passed!")

Best Practices

Quality Gates: ตั้ง Quality Gate ให้ Coverage > 80%, ไม่มี Critical Bugs
Domain Isolation: Domain Layer ต้องไม่ Import อะไรจาก Adapters
Port-first: ออกแบบ Port (Interface) ก่อน แล้วค่อยสร้าง Adapter
Test with Fakes: ใช้ In-memory Adapters สำหรับ Unit Tests เร็วและ Isolated
CI/CD Gate: รัน SonarQube ใน CI/CD Block Deploy ถ้าไม่ผ่าน Quality Gate
Dependency Rule: Dependencies ชี้เข้าใน ไม่ออกนอก Domain ไม่รู้จัก Infrastructure

SonarQube คืออะไร

Open-source Platform วิเคราะห์ Code Quality ตรวจจับ Bugs Code Smells Vulnerabilities Duplications รองรับ 30+ ภาษา Quality Gates มาตรฐาน CI/CD Community Edition ฟรี