Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_project_id(self):
self.call('python -m signac init my_project'.split())
assert str(signac.get_project()) == 'my_project'
assert self.call('python -m signac project'.split()).strip() == 'my_project'
def test_init_project_in_project_root(self):
self.call('python -m signac init my_project'.split())
assert str(signac.get_project()) == 'my_project'
with pytest.raises(ExitCodeError):
self.call('python -m signac init second_project'.split())
def make_project(alias='project', root=None, **kwargs):
"""Initialize a project for testing.
The initialized project has a few operations and a few jobs that are in
various points in the workflow defined by the project.
"""
init(alias=alias, root=root, template='testing')
project = signac.init_project(name=alias, root=root)
signac.testing.init_jobs(project, **kwargs)
return project
def test_sync_file(self):
self.call('python -m signac init ProjectA'.split())
project_a = signac.Project()
project_b = signac.init_project('ProjectB', os.path.join(self.tmpdir.name, 'b'))
job_src = project_a.open_job({'a': 0}).init()
job_dst = project_b.open_job({'a': 0}).init()
for i, job in enumerate([job_src, job_dst]):
with open(job.fn('test'), 'w') as file:
file.write('x'*(i+1))
# FileSyncConflict
with pytest.raises(ExitCodeError):
self.call('python -m signac sync {} {}'
.format(os.path.join(self.tmpdir.name, 'b'), self.tmpdir.name).split())
self.call('python -m signac sync {} {} --strategy never'
.format(os.path.join(self.tmpdir.name, 'b'), self.tmpdir.name).split())
with open(job_dst.fn('test'), 'r') as file:
assert file.read() == 'xx'
with pytest.raises(ExitCodeError):
self.call('python -m signac sync {} {}'
def test_get_job_nested_project(self):
# Test case: The job workspace dir is also a project root dir.
root = self._tmp_dir.name
project = signac.init_project(name='testproject', root=root)
job = project.open_job({'a': 1})
job.init()
with job:
nestedproject = signac.init_project('nestedproject')
nestedproject.open_job({'b': 2}).init()
assert project.get_job() == job
assert signac.get_job() == job
def test_config_show(self):
err = self.call('python -m signac config --local show'.split(), error=True).strip()
assert 'Did not find a local configuration file' in err
self.call('python -m signac init my_project'.split())
out = self.call('python -m signac config --local show'.split()).strip()
cfg = config.read_config_file('signac.rc')
expected = config.Config(cfg).write()
assert out.split(os.linesep) == expected
out = self.call('python -m signac config show'.split()).strip()
cfg = config.load_config()
expected = config.Config(cfg).write()
assert out.split(os.linesep) == expected
out = self.call('python -m signac config --global show'.split()).strip()
cfg = config.read_config_file(config.FN_CONFIG)
expected = config.Config(cfg).write()
assert out.split(os.linesep) == expected
def test_shell_with_jobs_and_selection_only_one_job(self):
self.call('python -m signac init my_project'.split())
project = signac.Project()
for i in range(3):
project.open_job(dict(a=i)).init()
assert len(project) == 3
out = self.call(
'python -m signac shell -f a 0',
'print(str(project), job, len(list(jobs))); exit()', shell=True)
job = list(project.find_jobs({'a': 0}))[0]
assert out.strip() == '>>> {} {} 1'.format(project, job)
def test_export(self):
self.call('python -m signac init my_project'.split())
project = signac.Project()
prefix_data = os.path.join(self.tmpdir.name, 'data')
err = self.call("python -m signac export {}"
.format(prefix_data).split(), error=True)
assert 'No jobs to export' in err
for i in range(10):
project.open_job({'a': i}).init()
assert len(project) == 10
self.call("python -m signac export {}".format(prefix_data).split())
assert len(project) == 10
assert len(os.listdir(prefix_data)) == 1
assert len(os.listdir(os.path.join(prefix_data, 'a'))) == 10
for i in range(10):
assert os.path.isdir(os.path.join(prefix_data, 'a', str(i)))
def test_import(self):
self.call('python -m signac init my_project'.split())
project = signac.Project()
prefix_data = os.path.join(self.tmpdir.name, 'data')
err = self.call("python -m signac import {}"
.format(self.tmpdir.name).split(), error=True)
assert 'Nothing to import.' in err
for i in range(10):
project.open_job({'a': i}).init()
job_ids = list(project.find_job_ids())
assert len(project) == 10
project.export_to(target=prefix_data, copytree=os.replace)
assert len(project) == 0
self.call("python -m signac import {}".format(prefix_data).split())
assert len(project) == 10
assert list(project.find_job_ids()) == job_ids
} for i in bad_chars]
view_prefix = os.path.join(self._tmp_pr, 'view')
for sp in statepoints:
self.project.open_job(sp).init()
with pytest.raises(RuntimeError):
self.project.create_linked_view(prefix=view_prefix)
class UpdateCacheAfterInitJob(signac.contrib.job.Job):
def init(self, *args, **kwargs):
super(UpdateCacheAfterInitJob, self).init(*args, **kwargs)
self._project.update_cache()
class UpdateCacheAfterInitJobProject(signac.Project):
"This is a test class that regularly calls the update_cache() method."
Job = UpdateCacheAfterInitJob
class TestCachedProject(TestProject):
project_class = UpdateCacheAfterInitJobProject
def test_repr(self):
repr(self)
class TestProjectInit():
@pytest.fixture(autouse=True)
def setUp(self, request):