<commit_before>import os
import random
import shutil
import string
import tempfile
import time
import unittest

from cxmanage.tftp import Tftp

class InternalTftpTest(unittest.TestCase):
    def setUp(self):
        self.work_dir = tempfile.mkdtemp()

        self.tftp = Tftp()
        self.tftp.set_internal_server(self.work_dir)

    def tearDown(self):
        self.tftp.kill_server()
        shutil.rmtree(self.work_dir)

    def test_put_and_get(self):
        # Create file
        contents = "".join([chr(random.randint(0, 255)) for a in range(1024)])
        filename = tempfile.mkstemp(prefix="%s/" % self.work_dir)[1]
        open(filename, "w").write(contents)

        # Upload
        basename = os.path.basename(filename)
        self.tftp.put_file(filename, basename)

        # Download
        self.tftp.get_file(basename, filename)

        # Verify match
        self.assertEqual(open(filename).read(), contents)

class ExternalTftpTest(unittest.TestCase):
    def setUp(self):
        self.work_dir = tempfile.mkdtemp()
        self.tftp_dir = tempfile.mkdtemp()

        # Set up an internal server
        self.internal_tftp = Tftp()
        self.internal_tftp.set_internal_server(self.tftp_dir)

        # Get address and port
        address = "localhost"
        port = self.internal_tftp.get_port()

        # Set up an external server
        self.tftp = Tftp()
        self.tftp.set_external_server(address, port)

    def tearDown(self):
        self.internal_tftp.kill_server()
        shutil.rmtree(self.tftp_dir)
        shutil.rmtree(self.work_dir)

    def test_put(self):
        # Create file
        contents = "".join([chr(random.randint(0, 255)) for a in range(1024)])
        filename = tempfile.mkstemp(prefix="%s/" % self.work_dir)[1]
        open(filename, "w").write(contents)

        # Upload
        basename = os.path.basename(filename)
        self.tftp.put_file(filename, basename)

        # Verify match
        self.assertEqual(contents,
                open("%s/%s" % (self.tftp_dir, basename)).read())

    def test_get(self):
        # Create file
        contents = "".join([chr(random.randint(0, 255)) for a in range(1024)])
        filename = tempfile.mkstemp(prefix="%s/" % self.tftp_dir)[1]
        open(filename, "w").write(contents)

        # Download
        basename = os.path.basename(filename)
        self.tftp.get_file(basename, "%s/%s" % (self.work_dir, basename))

        # Verify match
        self.assertEqual(contents,
                open("%s/%s" % (self.work_dir, basename)).read())

    def test_put_and_get(self):
        # Create file
        contents = "".join([chr(random.randint(0, 255)) for a in range(1024)])
        filename = tempfile.mkstemp(prefix="%s/" % self.work_dir)[1]
        open(filename, "w").write(contents)

        # Upload and remove original file
        basename = os.path.basename(filename)
        self.tftp.put_file(filename, basename)
        os.remove(filename)

        # Download
        self.tftp.get_file(basename, filename)

        # Verify match
        self.assertEqual(open(filename).read(), contents)<commit_msg>tests: Merge redundant TFTP tests<commit_after>import os
import random
import shutil
import string
import tempfile
import time
import unittest

from cxmanage.tftp import Tftp

class InternalTftpTest(unittest.TestCase):
    """ Tests involving an internal TFTP server """

    def setUp(self):
        self.work_dir = tempfile.mkdtemp()

        self.tftp = Tftp()
        self.tftp.set_internal_server(self.work_dir)

    def tearDown(self):
        self.tftp.kill_server()
        shutil.rmtree(self.work_dir)

    def test_put_and_get(self):
        """ Test file transfers on an internal host """

        # Create file
        contents = "".join([chr(random.randint(0, 255)) for a in range(1024)])
        filename = tempfile.mkstemp(prefix="%s/" % self.work_dir)[1]
        open(filename, "w").write(contents)

        # Upload
        basename = os.path.basename(filename)
        self.tftp.put_file(filename, basename)

        # Download
        self.tftp.get_file(basename, filename)

        # Verify match
        self.assertEqual(open(filename).read(), contents)

class ExternalTftpTest(unittest.TestCase):
    """ Tests involving an external TFTP server.

    For testing purposes the 'external' server points to an internally hosted
    one, but it allows us to make sure the actual TFTP protocol is working. """

    def setUp(self):
        self.work_dir = tempfile.mkdtemp()
        self.tftp_dir = tempfile.mkdtemp()

        # Set up an internal server
        self.internal_tftp = Tftp()
        self.internal_tftp.set_internal_server(self.tftp_dir)

        # Get address and port
        address = "localhost"
        port = self.internal_tftp.get_port()

        # Set up an external server
        self.tftp = Tftp()
        self.tftp.set_external_server(address, port)

    def tearDown(self):
        self.internal_tftp.kill_server()
        shutil.rmtree(self.tftp_dir)
        shutil.rmtree(self.work_dir)

    def test_put_and_get(self):
        """ Test file transfers on an external host """

        # Create file
        contents = "".join([chr(random.randint(0, 255)) for a in range(1024)])
        filename = tempfile.mkstemp(prefix="%s/" % self.work_dir)[1]
        open(filename, "w").write(contents)

        # Upload and remove original file
        basename = os.path.basename(filename)
        self.tftp.put_file(filename, basename)
        os.remove(filename)
        self.assertFalse(os.path.exists(filename))

        # Download
        self.tftp.get_file(basename, filename)

        # Verify match
        self.assertEqual(open(filename).read(), contents)
{'fraction_non_alphanumeric': 0.11562608998953611, 'fraction_numerical': 0.009417509591907918, 'mean_word_length': 2.593358395989975, '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': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6674858', 'n_tokens_mistral': 1897, 'n_tokens_neox': 1765, 'n_words': 371}