pytest + Hardware —
HW-in-the-loop test.

import pytest
import serial

# function scope: her test fonksiyonunda yeniden oluşturulur
@pytest.fixture(scope="function")
def tmp_dir(tmp_path):
    return tmp_path

# module scope: modül içindeki tüm testlerde tek örnek
@pytest.fixture(scope="module")
def serial_port():
    port = serial.Serial("/dev/ttyUSB0", baudrate=115200, timeout=2)
    yield port
    port.close()  # cleanup: test sonrası çalışır

# session scope: tüm test oturumunda tek örnek (bağlantı maliyeti yüksek kaynaklar)
@pytest.fixture(scope="session")
def jtag_connection():
    conn = OpenOCDSession("localhost", 4444)
    conn.connect()
    yield conn
    conn.disconnect()

import pytest

# Özel mark tanımı (pytest.ini veya pyproject.toml içinde kayıt gerekir)
# pytest.ini:
# [pytest]
# markers =
#     hil: hardware-in-the-loop testi
#     slow: uzun süren test
#     board_rpi4: Raspberry Pi 4 gerektirir

@pytest.mark.hil
@pytest.mark.board_rpi4
def test_uart_echo(serial_port):
    serial_port.write(b"PING\r\n")
    response = serial_port.read_until(b"\n")
    assert b"PONG" in response

# Çalıştırma:
# pytest -m hil          # sadece HIL testleri
# pytest -m "not slow"   # yavaş testler hariç

pip install pytest pytest-html pytest-timeout pytest-xdist
pip install pyserial          # UART
pip install gpiod             # GPIO (libgpiod Python binding)
pip install asyncserial       # async UART
pip install aiofiles          # async file I/O
pip install pytest-asyncio    # async test desteği

import pytest
import serial
import time
from typing import Optional

class SerialFixture:
    """UART bağlantısı ve yardımcı metodlar."""

    def __init__(self, port: str, baudrate: int = 115200, timeout: float = 5.0):
        self.port = serial.Serial(
            port=port,
            baudrate=baudrate,
            bytesize=serial.EIGHTBITS,
            parity=serial.PARITY_NONE,
            stopbits=serial.STOPBITS_ONE,
            timeout=timeout
        )

    def send_command(self, cmd: str, delay: float = 0.1) -> None:
        """Komut gönder ve kısa bekle."""
        self.port.write((cmd + "\r\n").encode())
        time.sleep(delay)

    def read_until(self, expected: str, timeout: float = 5.0) -> str:
        """Beklenen metni içeren satırı oku."""
        deadline = time.monotonic() + timeout
        buffer = ""
        while time.monotonic() < deadline:
            if self.port.in_waiting:
                buffer += self.port.read(self.port.in_waiting).decode(errors="replace")
                if expected in buffer:
                    return buffer
            time.sleep(0.01)
        raise TimeoutError(f"'{expected}' beklentisi zaman aşımına uğradı. Alınan: {buffer!r}")

    def flush(self) -> None:
        """Giriş tamponunu temizle."""
        self.port.reset_input_buffer()

    def close(self) -> None:
        self.port.close()


@pytest.fixture(scope="module")
def serial_dut(request):
    """Ana DUT UART bağlantısı (modül genelinde paylaşılır)."""
    port = request.config.getoption("--serial-port", default="/dev/ttyUSB0")
    baud = int(request.config.getoption("--baud", default="115200"))
    sf = SerialFixture(port, baudrate=baud)
    yield sf
    sf.close()


def pytest_addoption(parser):
    parser.addoption("--serial-port", default="/dev/ttyUSB0")
    parser.addoption("--baud", default="115200")

import pytest

def test_echo(serial_dut):
    """DUT'un echo komutunu doğru yanıtladığını doğrula."""
    serial_dut.flush()
    serial_dut.send_command("echo LAVA_PING")
    response = serial_dut.read_until("LAVA_PING")
    assert "LAVA_PING" in response

def test_kernel_version(serial_dut):
    serial_dut.flush()
    serial_dut.send_command("uname -r")
    response = serial_dut.read_until("#")
    assert "6." in response, f"Beklenen 6.x kernel, alınan: {response}"

@pytest.mark.parametrize("device", ["/dev/spi0.0", "/dev/i2c-1"])
def test_device_exists(serial_dut, device):
    serial_dut.flush()
    serial_dut.send_command(f"test -e {device} && echo EXIST || echo MISSING")
    response = serial_dut.read_until("EXIST\|MISSING")
    assert "EXIST" in response, f"{device} bulunamadı"

import telnetlib
import subprocess
import time
import pytest

class OpenOCDSession:
    """OpenOCD Telnet RPC arayüzü."""

    PROMPT = b"> "

    def __init__(self, host: str = "localhost", port: int = 4444):
        self.host = host
        self.port = port
        self._tn: telnetlib.Telnet = None

    def connect(self, timeout: float = 10.0) -> None:
        self._tn = telnetlib.Telnet(self.host, self.port, timeout=timeout)
        self._tn.read_until(self.PROMPT, timeout=timeout)

    def command(self, cmd: str, timeout: float = 10.0) -> str:
        """OpenOCD komutunu çalıştır ve yanıtı döndür."""
        self._tn.write((cmd + "\n").encode())
        response = self._tn.read_until(self.PROMPT, timeout=timeout)
        return response.decode(errors="replace").strip()

    def reset_halt(self) -> str:
        return self.command("reset halt")

    def resume(self) -> str:
        return self.command("resume")

    def flash_write(self, filepath: str, offset: int = 0) -> str:
        return self.command(f"flash write_image erase {filepath} {offset:#x}")

    def read_memory(self, address: int, count: int = 4) -> list[int]:
        """Bellek adresinden 32-bit kelimeler oku."""
        result = self.command(f"mdw {address:#x} {count}")
        # Çıktı: "0x20000000: deadbeef 00000001 ..."
        words = []
        for line in result.splitlines():
            if ":" in line:
                parts = line.split(":")[1].strip().split()
                words.extend(int(p, 16) for p in parts)
        return words

    def write_memory(self, address: int, value: int) -> str:
        return self.command(f"mww {address:#x} {value:#x}")

    def disconnect(self) -> None:
        if self._tn:
            self._tn.close()


@pytest.fixture(scope="session")
def openocd(request):
    """OpenOCD bağlantısı — oturum genelinde tek örnek."""
    config = request.config.getoption("--openocd-cfg", default="board/rpi4.cfg")
    # OpenOCD sürecini başlat
    proc = subprocess.Popen(
        ["openocd", "-f", config],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )
    time.sleep(2)  # OpenOCD'nin başlamasını bekle
    session = OpenOCDSession()
    session.connect()
    yield session
    session.disconnect()
    proc.terminate()

import pytest

MAGIC_ADDR = 0x20000000
MAGIC_VAL  = 0xDEADBEEF

def test_memory_write_read(openocd):
    """Bellek yaz/oku döngüsü testi."""
    openocd.reset_halt()
    openocd.write_memory(MAGIC_ADDR, MAGIC_VAL)
    values = openocd.read_memory(MAGIC_ADDR, 1)
    assert values[0] == MAGIC_VAL, \
        f"Beklenen {MAGIC_VAL:#x}, okunan {values[0]:#x}"
    openocd.resume()

def test_flash_and_boot(openocd, serial_dut, firmware_path):
    """Firmware yaz, boot et, seri konsol üzerinden doğrula."""
    openocd.reset_halt()
    result = openocd.flash_write(firmware_path)
    assert "wrote" in result.lower(), f"Flash yazma hatası: {result}"
    openocd.resume()

    # Boot mesajını bekle
    response = serial_dut.read_until("Boot complete", timeout=30)
    assert "Boot complete" in response

import gpiod
import time
import pytest

class GPIOFixture:
    """libgpiod Python binding üzerinden GPIO kontrolü."""

    def __init__(self, chip_path: str = "/dev/gpiochip0"):
        self.chip = gpiod.Chip(chip_path)
        self._lines: dict[int, gpiod.Line] = {}

    def request_output(self, pin: int, initial: int = 0) -> None:
        """Pin'i çıkış olarak yapılandır."""
        line = self.chip.get_line(pin)
        line.request(
            consumer="pytest-hil",
            type=gpiod.LINE_REQ_DIR_OUT,
            default_val=initial
        )
        self._lines[pin] = line

    def request_input(self, pin: int) -> None:
        """Pin'i giriş olarak yapılandır."""
        line = self.chip.get_line(pin)
        line.request(consumer="pytest-hil", type=gpiod.LINE_REQ_DIR_IN)
        self._lines[pin] = line

    def set(self, pin: int, value: int) -> None:
        self._lines[pin].set_value(value)

    def get(self, pin: int) -> int:
        return self._lines[pin].get_value()

    def toggle(self, pin: int) -> None:
        current = self.get(pin)
        self.set(pin, 1 - current)

    def pulse(self, pin: int, duration_ms: float = 100) -> None:
        """Pin'i belirtilen süre aktif et."""
        self.set(pin, 1)
        time.sleep(duration_ms / 1000)
        self.set(pin, 0)

    def wait_for_edge(self, pin: int, edge: str = "rising",
                      timeout_ms: float = 5000) -> bool:
        """Kenar tespitini bekle. True: kenar algılandı, False: timeout."""
        line = self.chip.get_line(pin)
        event_type = (
            gpiod.LINE_REQ_EV_RISING_EDGE
            if edge == "rising" else gpiod.LINE_REQ_EV_FALLING_EDGE
        )
        line.request(consumer="pytest-hil-edge", type=event_type)
        return line.event_wait(nsec=int(timeout_ms * 1_000_000))

    def release_all(self) -> None:
        for line in self._lines.values():
            line.release()
        self.chip.close()


@pytest.fixture(scope="module")
def gpio(request):
    chip = request.config.getoption("--gpio-chip", default="/dev/gpiochip0")
    g = GPIOFixture(chip)
    yield g
    g.release_all()

import pytest
import time

RESET_PIN = 17   # DUT reset pini (BCM)
LED_PIN   = 27   # LED çıkış pini
IRQ_PIN   = 22   # Interrupt giriş pini

def test_reset_pin_asserts(gpio, serial_dut):
    """Reset pini DUT'u yeniden başlatmalı."""
    gpio.request_output(RESET_PIN, initial=1)
    gpio.pulse(RESET_PIN, duration_ms=200)
    # DUT'un yeniden boot etmesini bekle
    response = serial_dut.read_until("login:", timeout=30)
    assert "login:" in response

def test_led_toggle(gpio):
    """LED pinini togglela ve geri oku."""
    gpio.request_output(LED_PIN, initial=0)
    gpio.set(LED_PIN, 1)
    assert gpio.get(LED_PIN) == 1
    gpio.set(LED_PIN, 0)
    assert gpio.get(LED_PIN) == 0

def test_interrupt_edge(gpio):
    """Giriş pininde yükselen kenar tespiti."""
    gpio.request_input(IRQ_PIN)
    # DUT interrupt üretmesi için komutu tetikle (ör. serial üzerinden)
    edge_detected = gpio.wait_for_edge(IRQ_PIN, "rising", timeout_ms=2000)
    assert edge_detected, "2 saniye içinde interrupt kenarı algılanmadı"

import pytest
import json
from pathlib import Path

# boards.json örnek içeriği:
# [
#   {"id": "rpi4-01", "serial": "/dev/ttyUSB0", "gpio_chip": "/dev/gpiochip0"},
#   {"id": "imx8-01", "serial": "/dev/ttyUSB1", "gpio_chip": "/dev/gpiochip1"}
# ]

def load_board_configs():
    cfg_file = Path(__file__).parent / "boards.json"
    if cfg_file.exists():
        return json.loads(cfg_file.read_text())
    return [{"id": "default", "serial": "/dev/ttyUSB0"}]


@pytest.fixture(params=load_board_configs(), ids=lambda b: b["id"])
def board_config(request):
    """Her board konfigürasyonu için fixture parametresi."""
    return request.param


@pytest.fixture
def serial_board(board_config):
    """board_config'den türetilen serial bağlantı."""
    sf = SerialFixture(board_config["serial"])
    yield sf
    sf.close()


def pytest_addoption(parser):
    parser.addoption(
        "--board", action="store", default=None,
        help="Belirli board ID'sini test et (örn. rpi4-01)"
    )

def pytest_collection_modifyitems(config, items):
    """--board seçeneği ile belirli board'u filtrele."""
    board_filter = config.getoption("--board")
    if not board_filter:
        return
    skip_marker = pytest.mark.skip(reason=f"--board={board_filter} filtresi")
    for item in items:
        if hasattr(item, "callspec"):
            board_id = item.callspec.params.get("board_config", {}).get("id")
            if board_id != board_filter:
                item.add_marker(skip_marker)

# Tüm board'ları test et
pytest tests/ --html=reports/multi-board.html --self-contained-html

# Paralel (birden fazla board aynı anda)
pytest tests/ -n 2 --html=reports/parallel.html

# Yalnızca belirli board
pytest tests/ --board=rpi4-01

# JUnit XML (GitLab CI için)
pytest tests/ --junit-xml=reports/junit.xml

import pytest
import asyncio
import pytest_asyncio

# pytest-asyncio konfigürasyonu
# pyproject.toml:
# [tool.pytest.ini_options]
# asyncio_mode = "auto"

@pytest_asyncio.fixture(scope="module")
async def async_serial():
    """asyncserial ile async UART bağlantısı."""
    # asyncserial: pip install asyncserial
    from asyncserial import AsyncSerial
    port = AsyncSerial("/dev/ttyUSB0", baudrate=115200)
    await port.open()
    yield port
    await port.close()


@pytest_asyncio.fixture(scope="session")
async def event_loop():
    """Tüm oturum için tek asyncio event loop."""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

import pytest
import asyncio

@pytest.mark.asyncio
async def test_concurrent_read(async_serial):
    """Eş zamanlı yazma ve okuma testi."""

    async def send_and_receive():
        await async_serial.write(b"PING\r\n")
        data = await asyncio.wait_for(
            async_serial.read_until(b"\n"),
            timeout=5.0
        )
        return data

    # İki eş zamanlı komut gönder
    result1, result2 = await asyncio.gather(
        send_and_receive(),
        asyncio.sleep(0.1)  # gecikme simülasyonu
    )
    assert b"PONG" in result1 or b"PING" in result1

@pytest.mark.asyncio
async def test_timeout_handling(async_serial):
    """Zaman aşımı doğru şekilde yönetilmeli."""
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(
            async_serial.read_until(b"NONEXISTENT_RESPONSE"),
            timeout=1.0
        )

[[runners]]
  name = "rpi4-hil-runner"
  url = "https://gitlab.example.com"
  token = "RUNNER_TOKEN"
  executor = "shell"
  # Donanıma erişim için shell executor kullanılır (Docker yerine)
  [runners.custom_build_dir]
  [runners.cache]

  # Tag ile HIL runner'ı işaretle
  # .gitlab-ci.yml içinde tags: [hil-board] ile hedefle

stages:
  - build
  - hil-test

build-firmware:
  stage: build
  image: registry.example.com/arm-toolchain:latest
  script:
    - cmake -B build -DCMAKE_TOOLCHAIN_FILE=arm-none-eabi.cmake
    - cmake --build build -j$(nproc)
  artifacts:
    paths:
      - build/firmware.elf
      - build/firmware.bin
    expire_in: 2h

hil-test-rpi4:
  stage: hil-test
  tags:
    - hil-board        # Bu runner tag'ine sahip runner'da çalışır
    - rpi4             # Spesifik board tipi
  needs: [build-firmware]
  resource_group: rpi4-01   # Board lock: aynı anda tek job
  variables:
    SERIAL_PORT: /dev/ttyUSB0
    GPIO_CHIP: /dev/gpiochip0
  script:
    - pip install -r tests/requirements.txt
    - |
      pytest tests/hil/ \
        --serial-port=$SERIAL_PORT \
        --gpio-chip=$GPIO_CHIP \
        --junit-xml=reports/junit.xml \
        --html=reports/hil-report.html \
        --self-contained-html \
        -v --timeout=120
  artifacts:
    reports:
      junit: reports/junit.xml
    paths:
      - reports/
    when: always
    expire_in: 1 week

hil-test-parallel:
  stage: hil-test
  tags: [hil-board]
  needs: [build-firmware]
  parallel:
    matrix:
      - BOARD_ID: [rpi4-01, rpi4-02, imx8-01]
  resource_group: ${BOARD_ID}
  script:
    - pytest tests/hil/ --board=$BOARD_ID --junit-xml=reports/${BOARD_ID}-junit.xml
  artifacts:
    reports:
      junit: reports/*-junit.xml
    when: always

"""HIL test conftest — serial + GPIO + power cycle."""
import pytest
import serial
import gpiod
import subprocess
import time
from dataclasses import dataclass

# ----- Board konfigürasyonu -----

@dataclass
class BoardConfig:
    serial_port: str
    baud: int
    gpio_chip: str
    reset_pin: int
    power_relay_script: str

DEFAULT_BOARD = BoardConfig(
    serial_port="/dev/ttyUSB0",
    baud=115200,
    gpio_chip="/dev/gpiochip0",
    reset_pin=17,
    power_relay_script="/usr/local/bin/power-relay.sh"
)

def pytest_addoption(parser):
    parser.addoption("--serial-port", default=DEFAULT_BOARD.serial_port)
    parser.addoption("--baud",         default=str(DEFAULT_BOARD.baud))
    parser.addoption("--gpio-chip",    default=DEFAULT_BOARD.gpio_chip)
    parser.addoption("--reset-pin",    default=str(DEFAULT_BOARD.reset_pin))

@pytest.fixture(scope="session")
def board(request):
    return BoardConfig(
        serial_port=request.config.getoption("--serial-port"),
        baud=int(request.config.getoption("--baud")),
        gpio_chip=request.config.getoption("--gpio-chip"),
        reset_pin=int(request.config.getoption("--reset-pin")),
        power_relay_script=DEFAULT_BOARD.power_relay_script
    )

# ----- Serial fixture -----

@pytest.fixture(scope="module")
def ser(board):
    port = serial.Serial(board.serial_port, board.baud, timeout=5)
    port.reset_input_buffer()
    yield port
    port.close()

# ----- GPIO fixture -----

@pytest.fixture(scope="module")
def gpio(board):
    chip = gpiod.Chip(board.gpio_chip)
    line = chip.get_line(board.reset_pin)
    line.request(consumer="pytest-hil", type=gpiod.LINE_REQ_DIR_OUT, default_val=1)
    yield line
    line.set_value(1)
    line.release()
    chip.close()

# ----- Power cycle yardımcısı -----

@pytest.fixture(scope="module")
def power_cycle(board):
    def _cycle(off_seconds: float = 3.0):
        subprocess.run([board.power_relay_script, "off"], check=True)
        time.sleep(off_seconds)
        subprocess.run([board.power_relay_script, "on"], check=True)
    return _cycle

"""Kapsamlı HIL test senaryoları."""
import pytest
import time

# --- UART Echo Testi ---

def test_uart_echo(ser):
    """UART'tan gönderilen verinin geri döneceğini doğrula."""
    test_str = b"HELLOWORLD"
    ser.reset_input_buffer()
    ser.write(test_str + b"\r\n")
    response = b""
    deadline = time.monotonic() + 5
    while time.monotonic() < deadline:
        response += ser.read(ser.in_waiting or 1)
        if test_str in response:
            break
    assert test_str in response, f"Echo alınamadı. Gelen: {response!r}"

# --- GPIO Reset Testi ---

def test_gpio_reset_reboots_dut(gpio, ser):
    """GPIO reset pini DUT'u yeniden başlatmalı."""
    # Reset pini 200 ms aşağı çek
    gpio.set_value(0)
    time.sleep(0.2)
    gpio.set_value(1)

    # Boot mesajını bekle
    ser.reset_input_buffer()
    buffer = b""
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline:
        buffer += ser.read(ser.in_waiting or 1)
        if b"login:" in buffer or b"Linux" in buffer:
            break
    assert b"Linux" in buffer or b"login:" in buffer, \
        "Reset sonrası boot başlamadı"

# --- Power Cycle Testi ---

@pytest.mark.slow
def test_power_cycle_clean_boot(power_cycle, ser):
    """Güç kesme/açma sonrası temiz boot testi."""
    power_cycle(off_seconds=5)

    ser.reset_input_buffer()
    buffer = b""
    deadline = time.monotonic() + 60
    while time.monotonic() < deadline:
        buffer += ser.read(ser.in_waiting or 1)
        if b"login:" in buffer:
            break

    assert b"login:" in buffer, \
        f"60 saniye içinde login prompt görülmedi. Alınan: {buffer[-500:]!r}"

# --- Entegrasyon: login + komut çalıştırma ---

@pytest.mark.hil
def test_full_login_and_command(ser):
    """Login yap, komut çalıştır, çıktıyı doğrula."""
    # Root login
    ser.write(b"root\r\n")
    time.sleep(0.5)

    # Komut gönder
    ser.reset_input_buffer()
    ser.write(b"cat /proc/version\r\n")
    time.sleep(1)

    output = ser.read(ser.in_waiting).decode(errors="replace")
    assert "Linux" in output, f"Kernel version alınamadı: {output!r}"