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_keras_log_weights(dummy_model, dummy_data, wandb_init_run):
dummy_model.fit(*dummy_data, epochs=2, batch_size=36, validation_data=dummy_data,
callbacks=[WandbCallback(data_type="image", log_weights=True)])
assert wandb_init_run.history.rows[0]['parameters/dense.weights']['_type'] == "histogram"
def test_keras_image_multiclass(dummy_model, dummy_data, wandb_init_run):
dummy_model.fit(*dummy_data, epochs=2, batch_size=36, validation_data=dummy_data,
callbacks=[WandbCallback(data_type="image", predictions=10)])
assert len(wandb_init_run.history.rows[0]["examples"]['captions']) == 10
def main():
wandb.init()
histogram_small_literal = wandb.Histogram(np_histogram=([1, 2, 4], [3, 10, 20, 0]))
histogram_large_random = wandb.Histogram(numpy.random.randint(255, size=(1000)))
numpy_array = numpy.random.rand(1000)
torch_tensor = torch.rand(1000, 1000)
data_frame = pandas.DataFrame(data=numpy.random.rand(1000), columns=['col'])
tensorflow_variable_single = tensorflow.Variable(543.01, tensorflow.float32)
tensorflow_variable_multi = tensorflow.Variable([[2, 3], [7, 11]], tensorflow.int32)
plot_scatter = plotly.graph_objs.Scatter(x=[0, 1, 2])
image_data = numpy.zeros((28, 28))
image_cool = wandb.Image(image_data, caption="Cool zeros")
image_nice = wandb.Image(image_data, caption="Nice zeros")
image_random = wandb.Image(numpy.random.randint(255, size=(28, 28, 3)))
image_pil = wandb.Image(PIL.Image.new("L", (28, 28)))
plt.plot([1, 2, 3, 4])
plt.ylabel('some interesting numbers')
image_matplotlib_plot = wandb.Image(plt)
matplotlib_plot = plt
audio_data = numpy.random.uniform(-1, 1, 44100)
sample_rate = 44100
caption1 = "This is what a dog sounds like"
caption2 = "This is what a chicken sounds like"
# test with all captions
audio1 = wandb.Audio(audio_data, sample_rate=sample_rate, caption=caption1)
audio2 = wandb.Audio(audio_data, sample_rate=sample_rate, caption=caption2)
# test with no captions
audio3 = wandb.Audio(audio_data, sample_rate=sample_rate)
def test_run_context_multi_run(live_mock_server, git_repo):
os.environ[env.BASE_URL] = "http://localhost:%i" % 8765
os.environ["WANDB_API_KEY"] = "B" * 40
with wandb.init() as run:
run.log({"a": 1, "b": 2})
with wandb.init(reinit=True) as run:
run.log({"c": 3, "d": 4})
assert len(glob.glob("wandb/*")) == 4
def test_restore_bad_remote(runner, request_mocker, query_run, git_repo, docker, monkeypatch):
# git_repo creates it's own isolated filesystem
mock = query_run(request_mocker, {"git": {"repo": "http://fake.git/foo/bar"}})
api = InternalApi({'project': 'test'})
monkeypatch.setattr(cli, 'api', api)
def bad_commit(cmt):
raise ValueError()
monkeypatch.setattr(api.git.repo, 'commit', bad_commit)
monkeypatch.setattr(api, "download_urls", lambda *args, **kwargs: [])
result = runner.invoke(cli.restore, ["wandb/test:abcdef"])
print(result.output)
print(traceback.print_tb(result.exc_info[2]))
assert result.exit_code == 1
assert "Run `git clone http://fake.git/foo/bar`" in result.output
def test_unwatch(wandb_init_run):
net = ConvNet()
wandb.watch(net, log_freq=1, log="all")
wandb.unwatch()
for i in range(3):
output = net(dummy_torch_tensor((64, 1, 28, 28)))
grads = torch.ones(64, 10)
output.backward(grads)
assert(len(wandb_init_run.history.row) == 0)
assert(
wandb_init_run.history.row.get('gradients/fc2.bias') is None)
wandb.log({"a": 2})
assert(len(wandb_init_run.history.rows) == 3)
def test_custom_file_policy_symlink(mocker, run_manager):
mod = mocker.MagicMock()
mocker.patch(
'wandb.run_manager.FileEventHandlerThrottledOverwriteMinWait.on_modified', mod)
with open("ckpt_0.txt", "w") as f:
f.write("joy")
with open("ckpt_1.txt", "w") as f:
f.write("joy" * 100)
wandb.save("ckpt_0.txt")
with open("ckpt_0.txt", "w") as f:
f.write("joy" * 100)
wandb.save("ckpt_1.txt")
run_manager.test_shutdown()
assert isinstance(
run_manager._file_event_handlers["ckpt_0.txt"], FileEventHandlerThrottledOverwriteMinWait)
assert mod.called
def test_save_absolute_path(wandb_init_run):
with open("/tmp/test.txt", "w") as f:
f.write("something")
wandb.save("/tmp/test.txt")
assert os.path.exists(os.path.join(wandb_init_run.dir, "test.txt"))
def test_retry_with_noauth_401(capsys):
def fail():
res = requests.Response()
res.status_code = 401
raise retry.TransientException(exc=requests.HTTPError(response=res))
fn = retry.Retry(fail, check_retry_fn=util.no_retry_auth)
with pytest.raises(CommError) as excinfo:
fn()
assert excinfo.value.message == 'Invalid or missing api_key. Run wandb login'
def test_captions():
wbone = wandb.Image(image, caption="Cool")
wbtwo = wandb.Image(image, caption="Nice")
assert wandb.Image.captions([wbone, wbtwo]) == ["Cool", "Nice"]