<commit_before>""" Holds state about the tftp service to be used by a calxeda update
application. """

import atexit, os, signal, shutil, subprocess

from cxfwupd_resources import cxfwupd_resources
from tftpy import TftpClient, TftpServer

class tftp:

    def __init__(self):
        self._ipaddr = None
        self._interface = None
        self._port = None

        self._isinternal = False
        self._hasbeengood = False

        self._server = None
        self._client = None

        atexit.register(self.kill_server)

    def get_settings_str(self):
        """ Get tftp settings as a string """
        strings = cxfwupd_resources.get_strings('tftp')
        s = ''
        if self._ipaddr == None:
            s = strings['not-set']
        else:
            s = self._ipaddr + ':' + str(self._port)
            if self._isinternal:
                s += strings['internal-server']
            if not self._hasbeengood:
                res = self.is_reachable()
            if res:
                s += strings['is-reachable']
            else:
                if self._hasbeengood:
                    s += strings['know-good-once']
                else:
                    s += strings['not-reachable']
        return s

    def is_set(self):
        """ Return true if a server has been set """
        return self._ipaddr != None

    def is_internal(self):
        """ Return true if we're using an internal server """
        return self._isinternal

    def is_reachable(self):
        """ Attempt to reach the tftp server. Return true if the server was
        reached successfully. """
        if self._ipaddr != None:
            # Try sending and receiving something small. if successful,
            # set _hasbeengood and return True; otherwise return False
            return True
        return False

    def set_internal_server(self, interface, port):
        """ Shut down any server that's currently running, then start an
        internal tftp server. """

        # Kill existing internal server
        self.kill_server()
<commit_msg>tftp: implemented is_reachable method<commit_after>""" Holds state about the tftp service to be used by a calxeda update
application. """

import atexit, os, signal, shutil, subprocess

from cxfwupd_resources import cxfwupd_resources
from tftpy import TftpClient, TftpServer

class tftp:

    def __init__(self):
        self._ipaddr = None
        self._interface = None
        self._port = None

        self._isinternal = False
        self._hasbeengood = False

        self._server = None
        self._client = None

        atexit.register(self.kill_server)

    def get_settings_str(self):
        """ Get tftp settings as a string """
        strings = cxfwupd_resources.get_strings('tftp')
        s = ''
        if self._ipaddr == None:
            s = strings['not-set']
        else:
            s = self._ipaddr + ':' + str(self._port)
            if self._isinternal:
                s += strings['internal-server']
            if not self._hasbeengood:
                res = self.is_reachable()
            if res:
                s += strings['is-reachable']
            else:
                if self._hasbeengood:
                    s += strings['know-good-once']
                else:
                    s += strings['not-reachable']
        return s

    def is_set(self):
        """ Return true if a server has been set """
        return self._ipaddr != None

    def is_internal(self):
        """ Return true if we're using an internal server """
        return self._isinternal

    def is_reachable(self):
        """ Attempt to reach the tftp server. Return true if the server was
        reached successfully. """
        if self._ipaddr != None:
            # Create a small file to send
            f = open("testfile_a", "w")
            f.write("")
            f.close()

            # Try sending and receiving the file
            try:
                self.put_file("testfile", "testfile_a")
                self.get_file("testfile", "testfile_b")
            except:
                pass

            # Check if we completed successfully
            os.remove("testfile_a")
            if os.path.exists("testfile_b"):
                os.remove("testfile_b")
                self._hasbeengood = True
                return True

        return False

    def set_internal_server(self, interface, port):
        """ Shut down any server that's currently running, then start an
        internal tftp server. """

        # Kill existing internal server
        self.kill_server()

{'fraction_non_alphanumeric': 0.08936355710549258, 'fraction_numerical': 0.0, 'mean_word_length': 1.7694628847314424, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '455848', 'n_tokens_mistral': 1274, 'n_tokens_neox': 1204, 'n_words': 378}