<commit_before>import os
import socket
import shutil
import tempfile
import pwd
from os import stat

import osgtest.library.core as core
import osgtest.library.files as files
import osgtest.library.osgunittest as osgunittest


class TestXrootd(osgunittest.OSGTestCase):

    __data_path = '/usr/share/osg-test/test_gridftp_data.txt'
    __fuse_path = '/mnt/xrootd_fuse_test'

    def test_01_xrdcp_local_to_server(self):
        core.skip_ok_unless_installed('xrootd', 'xrootd-client', by_dependency=True)
        if core.config['xrootd.gsi'] == "ON":
            core.skip_ok_unless_installed('globus-proxy-utils')
        self.skip_bad_unless(core.state['xrootd.started-server'] is True, 'Server not running')

        hostname = socket.getfqdn()
        if core.config['xrootd.gsi'] == "ON":
            temp_dir = "/tmp/vdttest"
            if not os.path.exists(temp_dir):
                os.mkdir(temp_dir)
                user = pwd.getpwnam(core.options.username)
                os.chown(temp_dir, user[2], user[3])
        else:
            temp_dir = tempfile.mkdtemp()
        os.chmod(temp_dir, 0o777)
        xrootd_url = 'root://%s/%s/copied_file.txt' % (hostname, temp_dir)
        command = ('xrdcp', '--debug', '3', TestXrootd.__data_path, xrootd_url)

        status, stdout, stderr = core.system(command, user=True)

        fail = core.diagnose('xrdcp copy, local to URL',
                             command, status, stdout, stderr)
        file_copied = os.path.exists(os.path.join(temp_dir, 'copied_file.txt'))
        shutil.rmtree(temp_dir)

        self.assertEqual(status, 0, fail)
        self.assert_(file_copied, 'Copied file missing')

    def test_02_test_multiuser(self):
        core.skip_ok_unless_installed('xrootd', 'xrootd-client', 'xrootd-multiuser', by_dependency=True)
        if core.config['xrootd.multiuser'] == "ON":
            vdttestpid = pwd.getpwnam("vdttest")
            owner = stat("/tmp/vdttest/copied_file.txt").st_uid
            self.assertEqual(owner, vdttestpid)

    def test_03_xrdcp_server_to_local(self):
        core.skip_ok_unless_installed('xrootd', 'xrootd-client', by_dependency=True)
        if core.config['xrootd.gsi'] == "ON":
            core.skip_ok_unless_installed('globus-proxy-utils')
        self.skip_bad_unless(core.state['xrootd.started-server'] is True, 'Server not running')

        hostname = socket.getfqdn()<commit_msg>Not removing the file we are testing for<commit_after>import os
import socket
import shutil
import tempfile
import pwd
from os import stat

import osgtest.library.core as core
import osgtest.library.files as files
import osgtest.library.osgunittest as osgunittest


class TestXrootd(osgunittest.OSGTestCase):

    __data_path = '/usr/share/osg-test/test_gridftp_data.txt'
    __fuse_path = '/mnt/xrootd_fuse_test'

    def test_01_xrdcp_local_to_server(self):
        core.skip_ok_unless_installed('xrootd', 'xrootd-client', by_dependency=True)
        if core.config['xrootd.gsi'] == "ON":
            core.skip_ok_unless_installed('globus-proxy-utils')
        self.skip_bad_unless(core.state['xrootd.started-server'] is True, 'Server not running')
        temp_dir = "/tmp/vdttest"
        hostname = socket.getfqdn()
        if core.config['xrootd.gsi'] == "ON":
            if not os.path.exists(temp_dir):
                os.mkdir(temp_dir)
                user = pwd.getpwnam(core.options.username)
                os.chown(temp_dir, user[2], user[3])
        else:
            temp_dir = tempfile.mkdtemp()
        os.chmod(temp_dir, 0o777)
        xrootd_url = 'root://%s/%s/copied_file.txt' % (hostname, temp_dir)
        command = ('xrdcp', '--debug', '3', TestXrootd.__data_path, xrootd_url)

        status, stdout, stderr = core.system(command, user=True)

        fail = core.diagnose('xrdcp copy, local to URL',
                             command, status, stdout, stderr)
        file_copied = os.path.exists(os.path.join(temp_dir, 'copied_file.txt'))
        if core.config['xrootd.multiuser'] != "ON":
            shutil.rmtree(temp_dir)
        self.assertEqual(status, 0, fail)
        self.assert_(file_copied, 'Copied file missing')

    def test_02_test_multiuser(self):
        core.skip_ok_unless_installed('xrootd', 'xrootd-client', 'xrootd-multiuser', by_dependency=True)
        temp_dir = "/tmp/vdttest"
        if core.config['xrootd.multiuser'] == "ON":
            vdttestpid = pwd.getpwnam("vdttest")
            owner = stat(os.path.join(temp_dir, 'copied_file.txt')).st_uid
            shutil.rmtree(temp_dir)
            self.assertEqual(owner, vdttestpid)

    def test_03_xrdcp_server_to_local(self):
        core.skip_ok_unless_installed('xrootd', 'xrootd-client', by_dependency=True)
        if core.config['xrootd.gsi'] == "ON":
            core.skip_ok_unless_installed('globus-proxy-utils')
        self.skip_bad_unless(core.state['xrootd.started-server'] is True, 'Server not running')

        hostname = socket.getfqdn()
{'fraction_non_alphanumeric': 0.1243471273603857, 'fraction_numerical': 0.00562474889513861, 'mean_word_length': 2.9359683794466402, 'pattern_counts': {'":': 9, '<': 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': '2992584', 'n_tokens_mistral': 1774, 'n_tokens_neox': 1701, 'n_words': 285}