<commit_before>import unittest
import tempfile
import shutil
import subprocess
import os, sys
import imp
import builtins
from io import StringIO

from os.path import join

from zeroinstall import SafeException
from zeroinstall.support import basedir
from zeroinstall.injector import qdom, model, gpg
from zeroinstall.injector.namespaces import XMLNS_IFACE

os.environ["http_proxy"] = "http://localhost:9999/bug"
mydir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.abspath('..'))

test_gpghome = join(mydir, 'test-gpghome')

from repo.cmd import main
from repo import archives, registry, paths, urltest

responses = {}		# Path -> Response

class FakeResponse:
	status = 200

	def __init__(self, size):
		self.size = size
		if size < 0:
			self.status = 404

	def getheader(self, name):
		assert name == 'Content-Length', name
		return str(self.size)

	def close(self):
		pass

class TestAPI:
	@staticmethod
	def upload(archives):
		for archive in archives:
			if 'INVALID' not in archive.rel_url:
				responses['/myrepo/archives/' + archive.rel_url] = FakeResponse(archive.size)

builtins.test0repo = TestAPI

class FakeHttpLib:
	class HTTPConnection:
		def __init__(self, host, port):
			pass

		def request(self, method, path, headers):
			assert method == 'HEAD'
			self.path = path

		def getresponse(self):
			return responses.get(self.path, FakeResponse(-1))

urltest.httplib = FakeHttpLib()
urltest.ftplib = None

gpg.ValidSig.is_trusted = lambda self, domain = None: True

def run_repo(args, stdin = ''):
	oldcwd = os.getcwd()

	old_stdout = sys.stdout
	sys.stdout = StringIO()
	try:
		sys.stdin = StringIO(stdin + '\n')	# (simulate pressing Return if needed)
		main(['0repo'] + args)
		return sys.stdout.getvalue()
	finally:
		os.chdir(oldcwd)
		sys.stdout = old_stdout

def update_config(old, new):
	if '0repo-config' in sys.modules:
		del sys.modules['0repo-config']

	with open('0repo-config.py') as stream:
		config_data = stream.read()

	with open('0repo-config.py', 'wt') as stream:
		new_data = config_data.replace(old, new)
		assert new_data != config_data
		stream.write(new_data)

	if os.path.exists('0repo-config.pyc'):
		os.unlink('0repo-config.pyc')

def test_invalid(xml):
	with open('test.xml', 'wt') as stream:
		stream.write(xml)
	try:
		run_repo(['add', 'test.xml'])
		assert 0, 'Not rejected'
	except SafeException as ex:
		return str(ex)

class Test0Repo(unittest.TestCase):
	def setUp(self):
		self.tmpdir = tempfile.mkdtemp('-0repo')
		os.chdir(self.tmpdir)
		gpghome = join(self.tmpdir, 'gnupg')
		shutil.copytree(test_gpghome, gpghome)
		os.environ['GNUPGHOME'] = gpghome
		os.chmod(gpghome, 0o700)

		os.environ['HOME'] = self.tmpdir

		for x in ['XDG_CONFIG_HOME', 'XDG_DATA_HOME', 'XDG_CACHE_HOME']:
			if x in os.environ:
				del os.environ[x]
		for x in ['XDG_CONFIG_DIRS', 'XDG_CACHE_DIRS', 'XDG_DATA_DIRS']:
			os.environ[x] = ''

		if 'ZEROINSTALL_PORTABLE_BASE' in os.environ:
			del os.environ['ZEROINSTALL_PORTABLE_BASE']

		imp.reload(basedir)

		responses.clear()

	def tearDown(self):
		if '0repo-config' in sys.modules:
			del sys.modules['0repo-config']
		os.chdir("/")
		shutil.rmtree(self.tmpdir)

	def testSimple(self):
		# (do a slow sub-process call here just to check that the top-level
		# wrapper works)
		subprocess.check_call(['0repo', 'create', 'my-repo', 'Test Key for 0repo'])
		os.chdir('my-repo')

		update_config('raise Exception("No upload method specified: edit upload_archives() in 0repo-config.py")',
				'return test0repo.upload(archives)')

		# Regenerate
		out = run_repo([])
		assert "Exported public key as 'public/keys/<KEY>'" in out, out
		assert os.path.exists(join('public', 'catalog.xml'))
		assert os.path.exists(join('public', 'resources/catalog.xsl'))
		assert os.path.exists(join('public', 'resources/catalog.css'))
		assert os.path.exists(join('public', 'keys', '<KEY>'))

		# Create a new feed (external archive)
		shutil.copyfile(join(mydir, 'test-1.xml'), join('incoming', 'test-1.xml'))
		responses['/downloads/test-1.tar.bz2'] = FakeResponse(419419)
		out = run_repo([])
		assert 'Processing test-1.xml' in out, repr(out)
<commit_msg>Fix unit-tests when source directory is read-only<commit_after>import unittest
import tempfile
import shutil
import subprocess
import os, sys
import imp
import builtins
from io import StringIO

from os.path import join

from zeroinstall import SafeException
from zeroinstall.support import basedir, ro_rmtree
from zeroinstall.injector import qdom, model, gpg
from zeroinstall.injector.namespaces import XMLNS_IFACE

os.environ["http_proxy"] = "http://localhost:9999/bug"
mydir = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.abspath('..'))

test_gpghome = join(mydir, 'test-gpghome')

from repo.cmd import main
from repo import archives, registry, paths, urltest

responses = {}		# Path -> Response

class FakeResponse:
	status = 200

	def __init__(self, size):
		self.size = size
		if size < 0:
			self.status = 404

	def getheader(self, name):
		assert name == 'Content-Length', name
		return str(self.size)

	def close(self):
		pass

class TestAPI:
	@staticmethod
	def upload(archives):
		for archive in archives:
			if 'INVALID' not in archive.rel_url:
				responses['/myrepo/archives/' + archive.rel_url] = FakeResponse(archive.size)

builtins.test0repo = TestAPI

class FakeHttpLib:
	class HTTPConnection:
		def __init__(self, host, port):
			pass

		def request(self, method, path, headers):
			assert method == 'HEAD'
			self.path = path

		def getresponse(self):
			return responses.get(self.path, FakeResponse(-1))

urltest.httplib = FakeHttpLib()
urltest.ftplib = None

gpg.ValidSig.is_trusted = lambda self, domain = None: True

def run_repo(args, stdin = ''):
	oldcwd = os.getcwd()

	old_stdout = sys.stdout
	sys.stdout = StringIO()
	try:
		sys.stdin = StringIO(stdin + '\n')	# (simulate pressing Return if needed)
		main(['0repo'] + args)
		return sys.stdout.getvalue()
	finally:
		os.chdir(oldcwd)
		sys.stdout = old_stdout

def update_config(old, new):
	if '0repo-config' in sys.modules:
		del sys.modules['0repo-config']

	with open('0repo-config.py') as stream:
		config_data = stream.read()

	with open('0repo-config.py', 'wt') as stream:
		new_data = config_data.replace(old, new)
		assert new_data != config_data
		stream.write(new_data)

	if os.path.exists('0repo-config.pyc'):
		os.unlink('0repo-config.pyc')

def test_invalid(xml):
	with open('test.xml', 'wt') as stream:
		stream.write(xml)
	try:
		run_repo(['add', 'test.xml'])
		assert 0, 'Not rejected'
	except SafeException as ex:
		return str(ex)

class Test0Repo(unittest.TestCase):
	def setUp(self):
		self.tmpdir = tempfile.mkdtemp('-0repo')
		os.chdir(self.tmpdir)
		gpghome = join(self.tmpdir, 'gnupg')
		shutil.copytree(test_gpghome, gpghome)
		os.environ['GNUPGHOME'] = gpghome
		os.chmod(gpghome, 0o700)

		os.environ['HOME'] = self.tmpdir

		for x in ['XDG_CONFIG_HOME', 'XDG_DATA_HOME', 'XDG_CACHE_HOME']:
			if x in os.environ:
				del os.environ[x]
		for x in ['XDG_CONFIG_DIRS', 'XDG_CACHE_DIRS', 'XDG_DATA_DIRS']:
			os.environ[x] = ''

		if 'ZEROINSTALL_PORTABLE_BASE' in os.environ:
			del os.environ['ZEROINSTALL_PORTABLE_BASE']

		imp.reload(basedir)

		responses.clear()

	def tearDown(self):
		if '0repo-config' in sys.modules:
			del sys.modules['0repo-config']
		os.chdir("/")
		ro_rmtree(self.tmpdir)

	def testSimple(self):
		# (do a slow sub-process call here just to check that the top-level
		# wrapper works)
		subprocess.check_call(['0repo', 'create', 'my-repo', 'Test Key for 0repo'])
		os.chdir('my-repo')

		update_config('raise Exception("No upload method specified: edit upload_archives() in 0repo-config.py")',
				'return test0repo.upload(archives)')

		# Regenerate
		out = run_repo([])
		assert "Exported public key as 'public/keys/<KEY>'" in out, out
		assert os.path.exists(join('public', 'catalog.xml'))
		assert os.path.exists(join('public', 'resources/catalog.xsl'))
		assert os.path.exists(join('public', 'resources/catalog.css'))
		assert os.path.exists(join('public', 'keys', '<KEY>'))

		# Create a new feed (external archive)
		shutil.copyfile(join(mydir, 'test-1.xml'), join('incoming', 'test-1.xml'))
		responses['/downloads/test-1.tar.bz2'] = FakeResponse(419419)
		out = run_repo([])
		assert 'Processing test-1.xml' in out, repr(out)

{'fraction_non_alphanumeric': 0.14030459287684374, 'fraction_numerical': 0.010792660990526441, 'mean_word_length': 4.541528239202658, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 16}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4396513', 'n_tokens_mistral': 3267, 'n_tokens_neox': 3012, 'n_words': 717}