<commit_before>"""
Database utility functions which use but are not actually models themselves
Used to allow models.py to be shortened and easier to follow.
"""

# pylint: disable=wrong-import-order

import os
import yaml
import jinja2
import datetime
import logging
import simplejson
from django.db.models import Q, Case, When, IntegerField, Sum
from django.db import IntegrityError, transaction
from django.contrib.auth.models import User
from django.utils import timezone
from linaro_django_xmlrpc.models import AuthToken
from lava_scheduler_app.models import (
    Device,
    DevicesUnavailableException,
    DeviceType,
    is_deprecated_json,
    JSONDataError,
    TestJob,
    TemporaryDevice,
    validate_job,
)
from lava_results_app.dbutils import map_metadata
from lava_dispatcher.pipeline.device import PipelineDevice

# pylint: disable=too-many-branches,too-many-statements,too-many-locals


def match_vlan_interface(device, job_def):
    if not isinstance(job_def, dict):
        raise RuntimeError("Invalid vlan interface data")
    if 'protocols' not in job_def or 'lava-vland' not in job_def['protocols'] or not device:
        return False
    interfaces = []
    logger = logging.getLogger('dispatcher-master')
    device_dict = device.load_configuration()
    if not device_dict or device_dict.get('parameters', {}).get('interfaces', None) is None:
        return False

    for vlan_name in job_def['protocols']['lava-vland']:
        tag_list = job_def['protocols']['lava-vland'][vlan_name]['tags']
        for interface in device_dict['parameters']['interfaces']:
            tags = device_dict['parameters']['interfaces'][interface]['tags']
            if not tags:
                continue
            logger.info(
                "Job requests %s for %s, device %s provides %s for %s",
                tag_list, vlan_name, device.hostname, tags, interface)
            if set(tags) & set(tag_list) == set(tag_list) and interface not in interfaces:
                logger.info("Matched vlan %s to interface %s on %s", vlan_name, interface, device)
                interfaces.append(interface)
                # matched, do not check any further interfaces of this device for this vlan
                break

    logger.info("Matched: %s", (len(interfaces) == len(job_def['protocols']['lava-vland'].keys())))
    return len(interfaces) == len(job_def['protocols']['lava-vland'].keys())


def initiate_health_check_job(device):
    logger = logging.getLogger('dispatcher-master')
    logger.info("Initiating health check")
    if not device:
        # logic error
        logger.error("No device")
        return None
    if device.status in [Device.RETIRED]:
        # logic error
        logger.error("[%s] has been retired", device)
        return None

    job_data = device.get_health_check()
    user = User.objects.get(username='lava-health')
    if not job_data:
        # This should never happen, it's a logic error.
        logger.error("No health check definition found for %s", device)
        device.put_into_maintenance_mode(
            user, "health check job not found in initiate_health_check_job")
        raise JSONDataError("no health check job found for %r", device.hostname)
    if is_deprecated_json(job_data):
        # only JSON supports 'target' and that needs to be set by the health-check not the admin.
        job_json = simplejson.loads(job_data)
        if 'target' in job_json:
            logger.error("[%s] JSON Health check definition must not specify a 'target'.", device.device_type.name)
            device.put_into_maintenance_mode(
                user, "target must not be defined in health check definitions.")
            return None
    try:
        job = testjob_submission(job_data, user, check_device=device)
    except DevicesUnavailableException as exc:
        logger.error("[%s] failed to submit health check - %s", device.device_type.name, exc)
        return None
    return job


def submit_health_check_jobs():
    """
    Checks which devices need a health check job and submits the needed
    health checks.
    Looping is only active once a device is offline.
    """

    logger = logging.getLogger('dispatcher-master')
    for device in Device.objects.filter(
            Q(status=Device.IDLE) | Q(status=Device.OFFLINE, health_status=Device.HEALTH_LOOPING)):
        time_denominator = True
        run_health_check = False
        if device.device_type.health_denominator == DeviceType.HEALTH_PER_JOB:
            time_denominator = False

        if device.device_type.disable_health_check:
            run_health_check = False<commit_msg>lava-server: except more exceptions<commit_after>"""
Database utility functions which use but are not actually models themselves
Used to allow models.py to be shortened and easier to follow.
"""

# pylint: disable=wrong-import-order

import os
import yaml
import jinja2
import datetime
import logging
import simplejson
from django.db.models import Q, Case, When, IntegerField, Sum
from django.db import IntegrityError, transaction
from django.contrib.auth.models import User
from django.utils import timezone
from linaro_django_xmlrpc.models import AuthToken
from lava_scheduler_app.models import (
    Device,
    DevicesUnavailableException,
    DeviceType,
    is_deprecated_json,
    JSONDataError,
    TestJob,
    TemporaryDevice,
    validate_job,
)
from lava_scheduler_app.schema import SubmissionException
from lava_results_app.dbutils import map_metadata
from lava_dispatcher.pipeline.device import PipelineDevice

# pylint: disable=too-many-branches,too-many-statements,too-many-locals


def match_vlan_interface(device, job_def):
    if not isinstance(job_def, dict):
        raise RuntimeError("Invalid vlan interface data")
    if 'protocols' not in job_def or 'lava-vland' not in job_def['protocols'] or not device:
        return False
    interfaces = []
    logger = logging.getLogger('dispatcher-master')
    device_dict = device.load_configuration()
    if not device_dict or device_dict.get('parameters', {}).get('interfaces', None) is None:
        return False

    for vlan_name in job_def['protocols']['lava-vland']:
        tag_list = job_def['protocols']['lava-vland'][vlan_name]['tags']
        for interface in device_dict['parameters']['interfaces']:
            tags = device_dict['parameters']['interfaces'][interface]['tags']
            if not tags:
                continue
            logger.info(
                "Job requests %s for %s, device %s provides %s for %s",
                tag_list, vlan_name, device.hostname, tags, interface)
            if set(tags) & set(tag_list) == set(tag_list) and interface not in interfaces:
                logger.info("Matched vlan %s to interface %s on %s", vlan_name, interface, device)
                interfaces.append(interface)
                # matched, do not check any further interfaces of this device for this vlan
                break

    logger.info("Matched: %s", (len(interfaces) == len(job_def['protocols']['lava-vland'].keys())))
    return len(interfaces) == len(job_def['protocols']['lava-vland'].keys())


def initiate_health_check_job(device):
    logger = logging.getLogger('dispatcher-master')
    logger.info("Initiating health check")
    if not device:
        # logic error
        logger.error("No device")
        return None
    if device.status in [Device.RETIRED]:
        # logic error
        logger.error("[%s] has been retired", device)
        return None

    job_data = device.get_health_check()
    user = User.objects.get(username='lava-health')
    if not job_data:
        # This should never happen, it's a logic error.
        logger.error("No health check definition found for %s", device)
        device.put_into_maintenance_mode(
            user, "health check job not found in initiate_health_check_job")
        raise JSONDataError("no health check job found for %r", device.hostname)
    if is_deprecated_json(job_data):
        # only JSON supports 'target' and that needs to be set by the health-check not the admin.
        job_json = simplejson.loads(job_data)
        if 'target' in job_json:
            logger.error("[%s] JSON Health check definition must not specify a 'target'.", device.device_type.name)
            device.put_into_maintenance_mode(
                user, "target must not be defined in health check definitions.")
            return None
    try:
        job = testjob_submission(job_data, user, check_device=device)
    except (DevicesUnavailableException, SubmissionException) as exc:
        logger.error("[%s] failed to submit health check - %s", device.device_type.name, exc)
        return None
    return job


def submit_health_check_jobs():
    """
    Checks which devices need a health check job and submits the needed
    health checks.
    Looping is only active once a device is offline.
    """

    logger = logging.getLogger('dispatcher-master')
    for device in Device.objects.filter(
            Q(status=Device.IDLE) | Q(status=Device.OFFLINE, health_status=Device.HEALTH_LOOPING)):
        time_denominator = True
        run_health_check = False
        if device.device_type.health_denominator == DeviceType.HEALTH_PER_JOB:
            time_denominator = False

        if device.device_type.disable_health_check:
            run_health_check = False
{'fraction_non_alphanumeric': 0.09163729573854534, 'fraction_numerical': 0.00021360674997329915, 'mean_word_length': 3.171046770601336, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1528576', 'n_tokens_mistral': 2667, 'n_tokens_neox': 2614, 'n_words': 783}