<commit_before>import unittest
import tempfile
import shutil
import os.path
import random

from gi.repository import Gio
from gi.repository import Gtk

import rename
import constants as c

import runtests


def _delete_recursively(gfile):
    if gfile.query_file_type(Gio.FileQueryInfoFlags.NOFOLLOW_SYMLINKS, None) == Gio.FileType.DIRECTORY:
        for fileinfo in gfile.enumerate_children(",".join([Gio.FILE_ATTRIBUTE_STANDARD_NAME, Gio.FILE_ATTRIBUTE_STANDARD_TYPE, Gio.FILE_ATTRIBUTE_STANDARD_IS_HIDDEN]), 0, None):
            _delete_recursively(gfile.get_child(fileinfo.get_name()))
    gfile.delete(None)


def _get_random_dir_name():
    with tempfile.TemporaryDirectory() as tmpdir:
        dirname = os.path.basename(tmpdir)
    return dirname


class _NameMap:
    def __init__(self, row, target_rel_dir=None):
        self.gfile_orig = row[c.FILES_MODEL_COLUMN_GFILE]
        if target_rel_dir is None:
            try:
                base_uri = row[c.FILES_MODEL_COLUMN_GFILE].get_parent().get_uri()
            except AttributeError:
                base_uri = ""
            self.gfile_prev = Gio.file_new_for_commandline_arg(base_uri+"/"+row[c.FILES_MODEL_COLUMN_PREVIEW])
        else:
            self.gfile_prev = target_rel_dir.resolve_relative_path(row[c.FILES_MODEL_COLUMN_PREVIEW])



class TestRenamer(unittest.TestCase):

    def __init__(self, *args):
        unittest.TestCase.__init__(self, *args)
        self.tmp_uri = None
        
        
    def tearDown(self):
        if False:
            print("Left tmp dir intact: {0}".format(self._tmp_dir.get_uri()))
        else:
            _delete_recursively(self._tmp_dir)

    def setUp(self):
        # tmp uri
        if not self.tmp_uri:
            self.tmp_uri = "file://" + tempfile.gettempdir() 
        if not self.tmp_uri.endswith("/"):
            self.tmp_uri = self.tmp_uri + "/"
        
        self._tmp_dir = Gio.file_new_for_commandline_arg(self.tmp_uri + _get_random_dir_name())
        try:
            self._tmp_dir.make_directory(None)
        except:
            self.fail("Base uri does not exist: {0}".format(self.tmp_uri))
        
        
        self._model = Gtk.ListStore(*c.FILES_MODEL_COLUMNS)
        self._mapping = None
        
        self._fail_msg = None
        
        self._two_pass = False
        

    def _add_to_model(self, model, original, preview, filepath):
        row = [None] * len(c.FILES_MODEL_COLUMNS)
        
        row[c.FILES_MODEL_COLUMN_ORIGINAL] = original
        row[c.FILES_MODEL_COLUMN_PREVIEW] = preview
        row[c.FILES_MODEL_COLUMN_GFILE] = filepath
        model.append(row)
        return row


    def _create_and_add_file_to_model(self, model, original, preview, rel_path=None):
        if rel_path is None:
            path = self._tmp_dir
        else:
            path = self._tmp_dir.resolve_relative_path(rel_path)
        filepath = path.resolve_relative_path(original)
        row = self._add_to_model(model, original, preview, filepath)
        
        ss = row[c.FILES_MODEL_COLUMN_GFILE].create(Gio.FileCreateFlags.REPLACE_DESTINATION, None)
        ss.write(row[c.FILES_MODEL_COLUMN_GFILE].get_uri().encode("utf-8"), None)
        ss.close(None)


    def _create_and_add_directory_to_model(self, model, original, preview, rel_path=None):
        if rel_path is None:
            path = self._tmp_dir
        else:
            path = self._tmp_dir.resolve_relative_path(rel_path)
        filepath = path.resolve_relative_path(original)
        row = self._add_to_model(model, original, preview, filepath)
        retval = row[c.FILES_MODEL_COLUMN_GFILE].make_directory_with_parents(None)
        self.assertTrue(retval)


    def _get_mapping(self, model):
        mapping = []
        for row in model:
            mapping.append(_NameMap(row))<commit_msg>Harden against sporadic remote failures<commit_after>import unittest
import tempfile
import shutil
import os.path
import random

from gi.repository import Gio
from gi.repository import Gtk

import rename
import constants as c

import runtests


def _delete_recursively(gfile):
    if gfile.query_file_type(Gio.FileQueryInfoFlags.NOFOLLOW_SYMLINKS, None) == Gio.FileType.DIRECTORY:

        success = False
        while not success:
            try:
                for fileinfo in gfile.enumerate_children(",".join([Gio.FILE_ATTRIBUTE_STANDARD_NAME, Gio.FILE_ATTRIBUTE_STANDARD_TYPE, Gio.FILE_ATTRIBUTE_STANDARD_IS_HIDDEN]), 0, None):
                    _delete_recursively(gfile.get_child(fileinfo.get_name()))
            except:
                pass
            else:
                success = True

    success = False
    while not success:
        try:
            gfile.delete(None)
        except:
            pass
        else:
            success = True


def _get_random_dir_name():
    with tempfile.TemporaryDirectory() as tmpdir:
        dirname = os.path.basename(tmpdir)
    return dirname


class _NameMap:
    def __init__(self, row, target_rel_dir=None):
        self.gfile_orig = row[c.FILES_MODEL_COLUMN_GFILE]
        if target_rel_dir is None:
            try:
                base_uri = row[c.FILES_MODEL_COLUMN_GFILE].get_parent().get_uri()
            except AttributeError:
                base_uri = ""
            self.gfile_prev = Gio.file_new_for_commandline_arg(base_uri+"/"+row[c.FILES_MODEL_COLUMN_PREVIEW])
        else:
            self.gfile_prev = target_rel_dir.resolve_relative_path(row[c.FILES_MODEL_COLUMN_PREVIEW])



class TestRenamer(unittest.TestCase):

    def __init__(self, *args):
        unittest.TestCase.__init__(self, *args)
        self.tmp_uri = None
    
    
    def tearDown(self):
        if False:
            print("Left tmp dir intact: {0}".format(self._tmp_dir.get_uri()))
        else:
            _delete_recursively(self._tmp_dir)

    def setUp(self):
        # tmp uri
        if not self.tmp_uri:
            self.tmp_uri = "file://" + tempfile.gettempdir() 
        if not self.tmp_uri.endswith("/"):
            self.tmp_uri = self.tmp_uri + "/"
        
        self._tmp_dir = Gio.file_new_for_commandline_arg(self.tmp_uri + _get_random_dir_name())
        
        assert self._tmp_dir.query_exists(None) == False
        
        while not self._tmp_dir.query_exists(None):
            try:
                self._tmp_dir.make_directory(None)
            except:
                pass
        
        self._model = Gtk.ListStore(*c.FILES_MODEL_COLUMNS)
        self._mapping = None
        
        self._fail_msg = None
        
        self._two_pass = False
        

    def _add_to_model(self, model, original, preview, filepath):
        row = [None] * len(c.FILES_MODEL_COLUMNS)
        
        row[c.FILES_MODEL_COLUMN_ORIGINAL] = original
        row[c.FILES_MODEL_COLUMN_PREVIEW] = preview
        row[c.FILES_MODEL_COLUMN_GFILE] = filepath
        model.append(row)
        return row


    def _create_and_add_file_to_model(self, model, original, preview, rel_path=None):
        if rel_path is None:
            path = self._tmp_dir
        else:
            path = self._tmp_dir.resolve_relative_path(rel_path)
        filepath = path.resolve_relative_path(original)
        row = self._add_to_model(model, original, preview, filepath)
        
        success = False
        while not success:
            try:
                ss = row[c.FILES_MODEL_COLUMN_GFILE].create(Gio.FileCreateFlags.REPLACE_DESTINATION, None)
            except:
                pass
            else:
                success = True
        
        ss.write(row[c.FILES_MODEL_COLUMN_GFILE].get_uri().encode("utf-8"), None)
        ss.close(None)


    def _create_and_add_directory_to_model(self, model, original, preview, rel_path=None):
        if rel_path is None:
            path = self._tmp_dir
        else:
            path = self._tmp_dir.resolve_relative_path(rel_path)
        filepath = path.resolve_relative_path(original)
        row = self._add_to_model(model, original, preview, filepath)
        success = False
        while not success:
            try:
                retval = row[c.FILES_MODEL_COLUMN_GFILE].make_directory_with_parents(None)
            except:
                pass
            else:
                success = True
        self.assertTrue(retval)


    def _get_mapping(self, model):
        mapping = []
        for row in model:
            mapping.append(_NameMap(row))
{'fraction_non_alphanumeric': 0.08831570188993225, 'fraction_numerical': 0.0008320456436467372, 'mean_word_length': 2.3125984251968505, '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': '5153780', 'n_tokens_mistral': 2743, 'n_tokens_neox': 2673, 'n_words': 468}