container_test() refactor + @hydazz ansi2html

[GilbN]

Fixes

• Made some of the test methods more failure proof.
• Made sure jinja autoescape is enabled when rendering the HTML template.

Adds

• Adds ansi2html from the @hydazz pr but with some changes.
• Add new option env: DOCKER_LOGS_DELAY - optional, How long to wait in seconds while tailing the container logs. Defaults to '300'
• Replace the old package dump with SBOM output.
• Adds a get_platform method
• Adds a @testing decorator that skips execution of a method if -e DRY_RUN=true is set.
  • Used on upload_file() and create_s3_client() methods.
• Adds color styling to python logs
  • 
• Adds a @deprecated decorator

Changes

• Splits up container_test() into multiple methods for readabilty.
• Correct exception logging.
• Readded the logblob object to the report_containers dictionary. (For the report.json file)
• Move all created files to be in /ci/output
• Use Ansi2HTMLConverter on the python logs
• Lock requirements.txt versions and use it in the Dockerfile
• Move creation of s3 client to a function
• Stops using the tester image for screenshots. Uses the chromedriver directly instead.

Logger:

• Muted urllib3 when DEBUG is true

Template:

• Fixed table styling
• Added a Pass/Fail header for each tag section
• Disabled anchor href when using the 404 image
• Update template for the new sbom output.

Test:

https://gilbnlsio2.s3.us-east-1.amazonaws.com/dockersabnzbd/PR29-test5/index.html https://gilbnlsio2.s3.us-east-1.amazonaws.com/dockersabnzbd/PR29-test4/index.html https://gilbnlsio2.s3.us-east-1.amazonaws.com/dockersabnzbd/PR29-test12/index.html

[hydazz]

The view buttons look iffy when viewed on mobile:

perhaps have view sbom output below log output?

[GilbN]

Yeah the sbom part is not finished. My testing overwrote the previous index file.

[hydazz]

is there a reason why a separate container with rdp is used rather than using selenium to directly get a screenshot locally?

ie:

    def take_screenshot(self, container: Container, tag: str) -> None:
        """Take a screenshot and save it to self.outdir

        Takes a screenshot using Selenium of the given endpoint and saves it to the specified tag in self.outdir.

        Args:
            `endpoint` (str): The endpoint to take a screenshot of.
            `tag` (str): The tag to use for the screenshot.
        """
        proto = 'https' if self.ssl.upper() == 'TRUE' else 'http'
        # Sleep for the user specified amount of time
        self.logger.info('Sleeping for %s seconds before reloading container: %s and refreshing container attrs',
                         self.test_container_delay, container.image)
        time.sleep(int(self.test_container_delay))
        container.reload()
        ip_adr = container.attrs['NetworkSettings']['Networks']['bridge']['IPAddress']
        endpoint = f'{proto}://{self.webauth}@{ip_adr}:{self.port}{self.webpath}'
        try:
            driver = self.setup_driver()
            driver.get(endpoint)
            self.logger.info(
                'Sleeping for %s seconds before creating a screenshot on %s', self.screenshot_delay, tag)
            time.sleep(int(self.screenshot_delay))
            self.logger.info('Taking screenshot of %s at %s', tag, endpoint)
            driver.get_screenshot_as_file(f'{tag}.png')
            # Compress and convert the screenshot to JPEG
            im = Image.open(f'{tag}.png').convert("RGB")
            im.save(f'{self.outdir}/{tag}.jpg', 'JPEG', quality=60)
            self.tag_report_tests[tag]['test']['Get screenshot'] = (dict(sorted({
                'status': 'PASS',
                'message': '-'}.items())))
            self.logger.info('Screenshot %s: PASS', tag)
        except (requests.Timeout, requests.ConnectionError, KeyError) as error:
            self.tag_report_tests[tag]['test']['Get screenshot'] = (dict(sorted({
                'status': 'FAIL',
                'message': f'CONNECTION ERROR: {error}'}.items())))
            self.logger.exception('Screenshot %s FAIL CONNECTION ERROR', tag)
        except TimeoutException as error:
            self.tag_report_tests[tag]['test']['Get screenshot'] = (dict(sorted({
                'status': 'FAIL',
                'message': f'TIMEOUT: {error}'}.items())))
            self.logger.exception('Screenshot %s FAIL TIMEOUT', tag)
        except (WebDriverException, Exception) as error:
            self.tag_report_tests[tag]['test']['Get screenshot'] = (dict(sorted({
                'status': 'FAIL',
                'message': f'UNKNOWN: {error}'}.items())))
            self.logger.exception('Screenshot %s FAIL UNKNOWN: %s', tag, error)
        finally:
            driver.quit()

    def setup_driver(self) -> webdriver.Chrome:
        """Return a single ChromiumDriver object the class can use

        Returns:
            Webdriver: Returns a Chromedriver object
        """
        self.logger.info("Init Chromedriver")
        # Selenium webdriver options
        chrome_options = webdriver.ChromeOptions()
        chrome_options.add_argument('--no-sandbox')
        chrome_options.add_argument('--headless')
        chrome_options.add_argument('--disable-gpu')
        chrome_options.add_argument('--window-size=1920x1080')
        chrome_options.add_argument('--disable-extensions')
        chrome_options.add_argument('--ignore-certificate-errors')
        # https://developers.google.com/web/tools/puppeteer/troubleshooting#tips
        chrome_options.add_argument('--disable-dev-shm-usage')
        driver = webdriver.Chrome(options=chrome_options)
        driver.set_page_load_timeout(60)
        return driver

produces:

I don't see any downsides, you can also get the response of the url this way, response != 200, fail, there is no chrome window, and it is still using chromium to render the webpage, so the pages should load correctly?

edit: I realize that chrome displays some error pages, however getting the response from selenium and phasing it seems more powerful

[GilbN]

Iirc the tester container helped out alot on the arm images and their qemu problems.

[hydazz]

does the syft container not need to have access to docker (docker.sock) otherwise it will try pull the container?

    def generate_sbom(self, tag: str) -> str:
        """Generate the SBOM for the image tag.

        Creates the output file in `{self.outdir}/{tag}.sbom.html`

        Args:
            tag (str): The tag we are testing

        Returns:
            bool: Return the output if successful otherwise "ERROR".
        """
        syft: Container = self.client.containers.run(
            image="anchore/syft:latest",
            command=f"{self.image}:{tag}",
            detach=True,
            volumes={
                '/var/run/docker.sock': {'bind': '/var/run/docker.sock', 'mode': 'rw'}
            }
        )

        self.logger.info('Creating SBOM package list on %s', tag)
        ...

2023/04/07 02:36:30 error during command execution: 1 error occurred:
    * failed to construct source from user input "<image_being_tested_locally>": could not fetch image "<image_being_tested_locally>": unable to use DockerDaemon source: pull failed: Error response from daemon: Head "<image_being_tested_locally>": denied

[hydazz]

also syft has added the requirement to specify the architecture when grabbing the sbom https://github.com/anchore/syft/issues/1708, otherwise it'll fail

(example fix)

        ...
        arch = ""
        if "amd64" in tag:
            arch = "amd64"
        elif "arm64" in tag:
            arch = "arm64"
        elif "arm32" in tag:
            arch = "arm32"

        syft: Container = self.client.containers.run(
            image="anchore/syft:latest",
            command=f"{self.image}:{tag} --platform linux/{arch}",
            ...

[GilbN]

also syft has added the requirement to specify the architecture when grabbing the sbom anchore/syft#1708, otherwise it'll fail

(example fix)

        ...
        arch = ""
        if "amd64" in tag:
            arch = "amd64"
        elif "arm64" in tag:
            arch = "arm64"
        elif "arm32" in tag:
            arch = "arm32"

        syft: Container = self.client.containers.run(
            image="anchore/syft:latest",
            command=f"{self.image}:{tag} --platform linux/{arch}",
            ...

Nice catch, I tested on the latest now and it didn't fail but I'm gonna add that.

And yes, the syft container needs docker sock, so my local testing has been flawed. The docker sock is mounted on the ci image, but that doesn't help the syft container.

[hydazz]

if your going to ansi2html the ci logs why not make each thread a different colour :smile::

logger.py

#!/usr/bin/env python3

import os
import logging
from logging.handlers import TimedRotatingFileHandler
import re
import platform
import sys

image = os.environ.get("IMAGE")
meta_tag = os.environ.get("META_TAG")
if image and meta_tag:
    dir = os.path.join(os.path.dirname(os.path.realpath(__file__)),"output",image,meta_tag)
    os.makedirs(dir, exist_ok=True)
    log_dir = os.path.join(dir,'ci.log')
else:
    log_dir = os.path.join(os.getcwd(),'ci.log')

logger = logging.getLogger()

class CustomLogFormatter(logging.Formatter):
    """Formatter that removes creds from logs."""
    ACCESS_KEY = os.environ.get("ACCESS_KEY","super_secret_key")
    SECRET_KEY = os.environ.get("SECRET_KEY","super_secret_key")

    # ANSI escape codes for different colors
    ANSI_CYAN = '\u001b[36m'
    ANSI_YELLOW = '\u001b[33m'
    ANSI_GREEN = '\u001b[32m'
    ANSI_RESET = '\u001b[0m'

    def formatException(self, exc_info):
        """Format an exception so that it prints on a single line."""
        result = super(CustomLogFormatter, self).formatException(exc_info)
        return repr(result)  # or format into one line however you want to

    def format_credential_key(self, s):
        return re.sub(self.ACCESS_KEY, '(removed)', s)

    def format_secret_key(self, s):
        return re.sub(self.SECRET_KEY, '(removed)', s)

    def format(self, record):
        s = super(CustomLogFormatter, self).format(record)
        if record.exc_text:
            s = s.replace('\n', '') + '|'
        s = self.format_credential_key(s)
        s = self.format_secret_key(s)

        # Set color based on thread name
        if "AMD64" in record.threadName:
            s = self.ANSI_CYAN + s + self.ANSI_RESET
        elif "ARM32" in record.threadName:
            s = self.ANSI_YELLOW + s + self.ANSI_RESET
        elif "ARM64" in record.threadName:
            s = self.ANSI_GREEN + s + self.ANSI_RESET

        return s

def configure_logging(log_level:str):
    """Setup console and file logging"""

    logger.handlers = []
    logger.setLevel(log_level)

    # Console logging
    ch = logging.StreamHandler(stream=sys.stdout)
    cf = CustomLogFormatter('%(asctime)-15s | %(threadName)-17s | %(name)-10s | %(levelname)-8s | (%(module)s.%(funcName)s|line:%(lineno)d) | %(message)s |', '%d/%m/%Y %H:%M:%S')
    ch.setFormatter(cf)
    ch.setLevel(log_level)
    logger.addHandler(ch)

    # File logging
    fh = TimedRotatingFileHandler(log_dir, when="midnight", interval=1, backupCount=7, delay=True, encoding='utf-8')
    f = CustomLogFormatter('%(asctime)-15s | %(threadName)-17s | %(name)-10s | %(levelname)-8s | (%(module)s.%(funcName)s|line:%(lineno)d) | %(message)s |', '%d/%m/%Y %H:%M:%S')
    fh.setFormatter(f)
    fh.setLevel(log_level)
    logger.addHandler(fh)

    logging.info('Operating system: %s', platform.platform())
    logging.info('Python version: %s', platform.python_version())
    if log_level.upper() == "DEBUG":
        logging.getLogger("botocore").setLevel(logging.WARNING) # Mute boto3 logging output
        logging.getLogger("urllib3.connectionpool").setLevel(logging.WARNING) # Mute urllib3.connectionpool logging output

[GilbN]

if your going to ansi2html the ci logs why not make each thread a different colour 😄:

logger.py

#!/usr/bin/env python3

import os
import logging
from logging.handlers import TimedRotatingFileHandler
import re
import platform
import sys

image = os.environ.get("IMAGE")
meta_tag = os.environ.get("META_TAG")
if image and meta_tag:
    dir = os.path.join(os.path.dirname(os.path.realpath(__file__)),"output",image,meta_tag)
    os.makedirs(dir, exist_ok=True)
    log_dir = os.path.join(dir,'ci.log')
else:
    log_dir = os.path.join(os.getcwd(),'ci.log')

logger = logging.getLogger()

class CustomLogFormatter(logging.Formatter):
    """Formatter that removes creds from logs."""
    ACCESS_KEY = os.environ.get("ACCESS_KEY","super_secret_key")
    SECRET_KEY = os.environ.get("SECRET_KEY","super_secret_key")

    # ANSI escape codes for different colors
    ANSI_CYAN = '\u001b[36m'
    ANSI_YELLOW = '\u001b[33m'
    ANSI_GREEN = '\u001b[32m'
    ANSI_RESET = '\u001b[0m'

    def formatException(self, exc_info):
        """Format an exception so that it prints on a single line."""
        result = super(CustomLogFormatter, self).formatException(exc_info)
        return repr(result)  # or format into one line however you want to

    def format_credential_key(self, s):
        return re.sub(self.ACCESS_KEY, '(removed)', s)

    def format_secret_key(self, s):
        return re.sub(self.SECRET_KEY, '(removed)', s)

    def format(self, record):
        s = super(CustomLogFormatter, self).format(record)
        if record.exc_text:
            s = s.replace('\n', '') + '|'
        s = self.format_credential_key(s)
        s = self.format_secret_key(s)

        # Set color based on thread name
        if "AMD64" in record.threadName:
            s = self.ANSI_CYAN + s + self.ANSI_RESET
        elif "ARM32" in record.threadName:
            s = self.ANSI_YELLOW + s + self.ANSI_RESET
        elif "ARM64" in record.threadName:
            s = self.ANSI_GREEN + s + self.ANSI_RESET

        return s

def configure_logging(log_level:str):
    """Setup console and file logging"""

    logger.handlers = []
    logger.setLevel(log_level)

    # Console logging
    ch = logging.StreamHandler(stream=sys.stdout)
    cf = CustomLogFormatter('%(asctime)-15s | %(threadName)-17s | %(name)-10s | %(levelname)-8s | (%(module)s.%(funcName)s|line:%(lineno)d) | %(message)s |', '%d/%m/%Y %H:%M:%S')
    ch.setFormatter(cf)
    ch.setLevel(log_level)
    logger.addHandler(ch)

    # File logging
    fh = TimedRotatingFileHandler(log_dir, when="midnight", interval=1, backupCount=7, delay=True, encoding='utf-8')
    f = CustomLogFormatter('%(asctime)-15s | %(threadName)-17s | %(name)-10s | %(levelname)-8s | (%(module)s.%(funcName)s|line:%(lineno)d) | %(message)s |', '%d/%m/%Y %H:%M:%S')
    fh.setFormatter(f)
    fh.setLevel(log_level)
    logger.addHandler(fh)

    logging.info('Operating system: %s', platform.platform())
    logging.info('Python version: %s', platform.python_version())
    if log_level.upper() == "DEBUG":
        logging.getLogger("botocore").setLevel(logging.WARNING) # Mute boto3 logging output
        logging.getLogger("urllib3.connectionpool").setLevel(logging.WARNING) # Mute urllib3.connectionpool logging output

Not something that we will benefit much from as its jenkins that runs it :) But I do use it on a work project based on the log level. 👍

[hydazz]

Not something that we will benefit much from as its jenkins that runs it :) But I do use it on a work project based on the log level. 👍

Colours are rendered in blueocean and in python.log.html, just a kool addition haha:

[GilbN]

Not something that we will benefit much from as its jenkins that runs it :) But I do use it on a work project based on the log level. 👍

Colours are rendered in blueocean and in python.log.html, just a kool addition haha:

OH, well that's cool.

[GilbN]

https://gilbnlsio2.s3.us-east-1.amazonaws.com/dockersabnzbd/PR29-test9/index.html

[GilbN]

Use the chrome driver directly instead of the tester container.

Test: https://gilbnlsio2.s3.us-east-1.amazonaws.com/dockersabnzbd/PR29-test12/index.html