<commit_before>#!/usr/bin/python3
# pylint: disable=wrong-import-position,too-many-locals,unused-argument,wrong-import-order
#
# Copyright (C) 2017 <NAME> <<EMAIL>>
#
# SPDX-License-Identifier: LGPL-2.1+

import gi
import os
import requests
import time
import sys
import glob
import json
from termcolor import colored

gi.require_version("Fwupd", "2.0")

from gi.repository import Fwupd
from gi.repository import Gio
from gi.repository import GLib


def _get_cache_file(fn):
    cachedir = os.path.expanduser("~/.cache/fwupdmgr")
    if not os.path.exists(cachedir):
        os.makedirs(cachedir)
    cachefn = os.path.join(cachedir, fn)
    if not os.path.exists(cachefn):
        url = "https://fwupd.org/downloads/" + fn
        print("Downloading", url)
        r = requests.get(url)
        r.raise_for_status()
        f = open(cachefn, "wb")
        f.write(r.content)
        f.close()
    return cachefn


class DeviceTest:
    def _print_request(self, req):
        self._info("ID: {}".format(req.get_id()))
        self._info("Message: {}".format(req.get_message()))

    def _device_request_cb(self, client, req):
        if req.get_kind() == Fwupd.RequestKind.IMMEDIATE:
            self._print_request(req)
            return
        if req.get_kind() == Fwupd.RequestKind.POST:
            self._request_queue.append(req)

    def _percentage_cb(self, client, _gparamstring):
        self._info("Percentage: {}".format(client.get_percentage()))

    def _status_cb(self, client, _gparamstring):
        self._info("Status: {}".format(Fwupd.status_to_string(client.get_status())))

    def __init__(self, obj):
        self.client = Fwupd.Client.new()
        self.name = obj.get("name", "Unknown")
        self.guids = obj.get("guids", [])
        self.releases = obj.get("releases", [])
        self.has_runtime = obj.get("runtime", True)
        self.interactive = obj.get("interactive", False)
        self.disabled = obj.get("disabled", False)
        self.protocol = obj.get("protocol", None)
        self._request_queue = []

        # use a custom main context
        main_ctx = GLib.MainContext.new()
        self.client.set_main_context(main_ctx)

        # support DeviceRequest
        self.client.set_feature_flags(Fwupd.FeatureFlags.REQUESTS)
        self.client.connect("device-request", self._device_request_cb)
        self.client.connect("notify::percentage", self._percentage_cb)
        self.client.connect("notify::status", self._status_cb)

    def _info(self, msg):
        print(colored("[INFO]".ljust(10), "blue"), msg)

    def _warn(self, msg):
        print(colored("[WARN]".ljust(10), "yellow"), msg)

    def _failed(self, msg):
        print(colored("[FAILED]".ljust(10), "red"), msg)

    def _success(self, msg):
        print(colored("[SUCCESS]".ljust(10), "green"), msg)

    def _get_by_device_guids(self):
        cancellable = Gio.Cancellable.new()
        for d in self.client.get_devices(cancellable):
            for guid in self.guids:
                if d.has_guid(guid):
                    if self.protocol and self.protocol != d.get_protocol():
                        continue
                    return d
        return None

    def run(self):

        print("Running test on {}".format(self.name))
        dev = self._get_by_device_guids()
        if not dev:
            self._warn("no {} attached".format(self.name))
            return

        self._info("Current version {}".format(dev.get_version()))

        # apply each file
        for obj in self.releases:<commit_msg>device-tests: Do not use a deprecated method<commit_after>#!/usr/bin/python3
# pylint: disable=wrong-import-position,too-many-locals,unused-argument,wrong-import-order
#
# Copyright (C) 2017 <NAME> <<EMAIL>>
#
# SPDX-License-Identifier: LGPL-2.1+

import gi
import os
import requests
import time
import sys
import glob
import json
from termcolor import colored

gi.require_version("Fwupd", "2.0")

from gi.repository import Fwupd
from gi.repository import Gio
from gi.repository import GLib


def _get_cache_file(fn):
    cachedir = os.path.expanduser("~/.cache/fwupdmgr")
    if not os.path.exists(cachedir):
        os.makedirs(cachedir)
    cachefn = os.path.join(cachedir, fn)
    if not os.path.exists(cachefn):
        url = "https://fwupd.org/downloads/" + fn
        print("Downloading", url)
        r = requests.get(url)
        r.raise_for_status()
        f = open(cachefn, "wb")
        f.write(r.content)
        f.close()
    return cachefn


class DeviceTest:
    def _print_request(self, req):
        self._info("ID: {}".format(req.get_id()))
        self._info("Message: {}".format(req.get_message()))

    def _device_request_cb(self, client, req):
        if req.get_kind() == Fwupd.RequestKind.IMMEDIATE:
            self._print_request(req)
            return
        if req.get_kind() == Fwupd.RequestKind.POST:
            self._request_queue.append(req)

    def _percentage_cb(self, client, _gparamstring):
        self._info("Percentage: {}".format(client.get_percentage()))

    def _status_cb(self, client, _gparamstring):
        self._info("Status: {}".format(Fwupd.status_to_string(client.get_status())))

    def __init__(self, obj):
        self.client = Fwupd.Client.new()
        self.name = obj.get("name", "Unknown")
        self.guids = obj.get("guids", [])
        self.releases = obj.get("releases", [])
        self.has_runtime = obj.get("runtime", True)
        self.interactive = obj.get("interactive", False)
        self.disabled = obj.get("disabled", False)
        self.protocol = obj.get("protocol", None)
        self._request_queue = []

        # use a custom main context
        main_ctx = GLib.MainContext.new()
        self.client.set_main_context(main_ctx)

        # support DeviceRequest
        self.client.set_feature_flags(Fwupd.FeatureFlags.REQUESTS)
        self.client.connect("device-request", self._device_request_cb)
        self.client.connect("notify::percentage", self._percentage_cb)
        self.client.connect("notify::status", self._status_cb)

    def _info(self, msg):
        print(colored("[INFO]".ljust(10), "blue"), msg)

    def _warn(self, msg):
        print(colored("[WARN]".ljust(10), "yellow"), msg)

    def _failed(self, msg):
        print(colored("[FAILED]".ljust(10), "red"), msg)

    def _success(self, msg):
        print(colored("[SUCCESS]".ljust(10), "green"), msg)

    def _get_by_device_guids(self):
        cancellable = Gio.Cancellable.new()
        for d in self.client.get_devices(cancellable):
            for guid in self.guids:
                if d.has_guid(guid):
                    if self.protocol and not d.has_protocol(self.protocol):
                        continue
                    return d
        return None

    def run(self):

        print("Running test on {}".format(self.name))
        dev = self._get_by_device_guids()
        if not dev:
            self._warn("no {} attached".format(self.name))
            return

        self._info("Current version {}".format(dev.get_version()))

        # apply each file
        for obj in self.releases:
{'fraction_non_alphanumeric': 0.1379794857383729, 'fraction_numerical': 0.00477729380356892, 'mean_word_length': 2.6823590274185203, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5873901', 'n_tokens_mistral': 2312, 'n_tokens_neox': 2246, 'n_words': 443}