test: wait for Redis readiness in the EURECOM lab

import json

from pathlib import Path

import sys

import subprocess

from types import SimpleNamespace

import pytest

ROOT=Path(__file__).resolve().parents[3]

sys.path.insert(0,str(ROOT/'ci'))

sys.path.insert(0,str(ROOT/'ansible/tests/r2_e2e'))

from eurecom_lab import SERVICES, composition

from eurecom_lab import SERVICES, composition, EurecomLab

from provider_observer import read_subscriptions, ti_collection

import provider_observer

        assert service['networks']=={'lab':{'aliases':[name]}}

        assert service['volumes']==['/owned/shared/config/service.yaml:/etc/config.yaml:ro']

        assert service['environment']==source['services'][name]['environment']

        if name!='redis':

            assert service['healthcheck']==source['services'][name]['healthcheck']

def test_ti_collection_error_or_malformed_body_cannot_prove_absence(status,body):

    with pytest.raises(AssertionError):

        ti_collection(SimpleNamespace(status_code=status,json=lambda:body))

def test_redis_consumers_wait_for_ready_service_without_removing_other_dependencies():

    source=original()

    source['services']['core-network-service']['depends_on']={

        'core-simulator':{'condition':'service_healthy'},'redis':{'condition':'service_started'}}

    actual=composition(source,{'upstream:test':'example/nef@sha256:'+'a'*64},'k3d-owned','/owned','oop-ci-owned')['services']

    assert actual['redis']['healthcheck']['test']==['CMD','redis-cli','ping']

    for name in ('ue-identity','ue-profile','monitoring-event','core-network-service'):

        assert actual[name]['depends_on']['redis']=={'condition':'service_healthy'}

    assert actual['core-network-service']['depends_on']['core-simulator']=={'condition':'service_healthy'}

def test_failed_start_records_safe_states_before_cleanup_and_preserves_failure(tmp_path,capsys):

    failure=subprocess.CalledProcessError(1,['docker','compose','up'])

    def call(command,**kwargs):

        if 'up' in command:raise failure

        if 'inspect' in command:return SimpleNamespace(stdout=json.dumps([{'State':{

            'Status':'exited','Running':False,'ExitCode':1,'OOMKilled':False,

            'Error':'PRIVATE-CONFIG','Health':{'Status':'unhealthy','Log':['PRIVATE-CONFIG']}}}]))

        return SimpleNamespace(stdout='Redis:6379 connection refused PRIVATE-CONFIG',stderr='')

    lab=EurecomLab('owned',tmp_path,'owned',call);lab.compose=['docker','compose']

    with pytest.raises(subprocess.CalledProcessError) as caught:lab._start_services()

    assert caught.value is failure

    states=lab.evidence['startup_failure']

    assert states['ue-identity']=={'status':'exited','exit_code':1,'oom_killed':False,

                                 'health':'unhealthy','redis_connection_error':True}

    assert 'PRIVATE-CONFIG' not in json.dumps(lab.evidence)+capsys.readouterr().out

import json

from pathlib import Path

import re

import subprocess

import tarfile

import time

from urllib.request import urlopen

            service['volumes'] = [str(Path(directory) / v.split(':')[0])+':'+v.split(':')[1]+':ro'

                                  for v in service['volumes']]

        services[key] = service

    # Upstream middleware connects to Redis during startup. A container start

    # alone does not prove that the Redis modules and socket are ready.

    services['redis']['healthcheck'] = {

        'test':['CMD','redis-cli','ping'], 'interval':'2s', 'timeout':'2s', 'retries':30,

    }

    for key in ('ue-identity', 'ue-profile', 'monitoring-event', 'core-network-service'):

        services[key].setdefault('depends_on', {})['redis'] = {'condition':'service_healthy'}

    return {'services':services, 'networks':{'lab':{'external':True,'name':network}}}

        path.write_text(yaml.safe_dump(config,sort_keys=False))

        self.compose = ['docker','compose','-f',str(path),'-p',self.name]

        self.owned = True  # Clean even partial startup.

        self.call([*self.compose,'up','-d','--wait','--wait-timeout','180'],timeout=360)

        self._start_services()

        def address(service):

            info = json.loads(self.call(['docker','inspect',self.name+'-'+service],capture_output=True).stdout)[0]

            return info['NetworkSettings']['Networks'][self.network]['IPAddress']

        self.evidence.update(startup='passed', device_ipv4=self.device_ip, base_url=self.base_url)

        return self

    def _start_services(self):

        try:

            self.call([*self.compose,'up','-d','--wait','--wait-timeout','180'],timeout=360)

        except subprocess.SubprocessError:

            # Retain only safe state/error classifications before owned cleanup.

            # Raw logs can contain configuration; they never enter CI artifacts.

            failures = {}

            for key in SERVICES:

                try:

                    info = json.loads(self.call(['docker','inspect',self.name+'-'+key],

                                               capture_output=True).stdout)[0]['State']

                    if info.get('Running') and info.get('Health',{}).get('Status') == 'healthy':

                        continue

                    logs = self.call(['docker','logs','--tail','100',self.name+'-'+key],

                                     capture_output=True)

                    failures[key] = {

                        'status':info.get('Status'), 'exit_code':info.get('ExitCode'),

                        'oom_killed':info.get('OOMKilled'),

                        'health':info.get('Health',{}).get('Status'),

                        'redis_connection_error':bool(re.search(

                            r'(?i)(redis|:6379).*?(refused|timeout|no such host|failed|error)',

                            logs.stdout + logs.stderr)),

                    }

                except (OSError, ValueError, KeyError, IndexError, subprocess.SubprocessError):

                    failures[key] = {'diagnostic':'unavailable'}

            self.evidence['startup_failure'] = failures

            print('Provider startup state: '+json.dumps(failures), flush=True)

            raise

    def close(self):

        if self.owned:

            self.call([*self.compose,'down','--volumes','--remove-orphans'],timeout=180)

smoke-command tests exercise the actual Ansible/pytest path against controlled

HTTP endpoints. A local DinD proof does not replace execution on the selected

GitLab runner or qualification of release-specific images.

EURECOM middleware waits for a successful Redis PING before starting. The lab

retains upstream health checks and the CoreSim dependency. A failed provider

startup records only container state, exit code, OOM status, health and a Redis

connection-error classification before cleanup; raw provider logs stay private.