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_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):
def test_init(self):
root = self._tmp_dir.name
with pytest.raises(LookupError):
signac.get_project(root=root)
project = signac.init_project(name='testproject', root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
# Second initialization should not make any difference.
project = signac.init_project(name='testproject', root=root)
project = signac.get_project(root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
project = signac.Project.get_project(root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
# Deviating initialization parameters should result in errors.
with pytest.raises(RuntimeError):
signac.init_project(name='testproject2', root=root)
with pytest.raises(RuntimeError):
signac.init_project(
name='testproject',
root=root,
workspace='workspace2')
with pytest.raises(RuntimeError):
signac.init_project(
name='testproject2',
root=root,
workspace='workspace2')
def test_clone(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 = project_a.open_job({'a': 0})
job.init()
assert len(project_a) == 1
assert len(project_b) == 0
self.call("python -m signac clone {} {}"
.format(os.path.join(self.tmpdir.name, 'b'), job.id).split())
assert len(project_a) == 1
assert job in project_a
assert len(project_b) == 1
assert job in project_b
# cloning a job that exist at both source and destination
err = self.call("python -m signac clone {} {}"
.format(os.path.join(self.tmpdir.name, 'b'), job.id).split(), error=True)
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):
def test_index(self):
self.call('python -m signac init my_project'.split())
self.call('python -m signac project --access'.split())
project = signac.Project()
project.open_job({'a': 0}).init()
assert len(project) == 1
with pytest.deprecated_call():
assert len(list(project.index())) == 1
assert len(list(signac.index())) == 1
doc = json.loads(self.call('python -m signac index'.split()))
assert 'statepoint' in doc
assert doc['statepoint'] == {'a': 0}
doc = json.loads(self.call('python -m signac project --index'.split()))
assert 'statepoint' in doc
assert doc['statepoint'] == {'a': 0}
project.open_job({'a': 0}).document['b'] = 0
doc = json.loads(self.call('python -m signac index'.split()))
assert 'statepoint' in doc
assert doc['statepoint'] == {'a': 0}
assert 'b' in doc
# Copyright (c) 2019 The Regents of the University of Michigan
# All rights reserved.
# This software is licensed under the BSD 3-Clause License.
import pytest
import signac
from tempfile import TemporaryDirectory
class TestDiffBase():
project_class = signac.Project
@pytest.fixture(autouse=True)
def setUp(self, request):
self._tmp_dir = TemporaryDirectory(prefix='signac_')
request.addfinalizer(self._tmp_dir.cleanup)
self.project = self.project_class.init_project(
name='diff_test_project',
root=self._tmp_dir.name)
class TestDiff(TestDiffBase):
def test_two_jobs(self):
job1 = self.project.open_job({'a': 0, 'b': 1})
job2 = self.project.open_job({'a': 0})
expected = {str(job1.id): {'b': 1}, str(job2.id): {}}
def test_get_project(self):
root = self._tmp_dir.name
with pytest.raises(LookupError):
signac.get_project(root=root)
project = signac.init_project(name='testproject', root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
project = signac.Project.init_project(name='testproject', root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
project = signac.get_project(root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root
project = signac.Project.get_project(root=root)
assert project.id == 'testproject'
assert project.workspace() == os.path.join(root, 'workspace')
assert project.root_directory() == root