import os
import tempfile
import shutil
from dotenv import load_dotenv
from setuptools import setup, find_packages
load_dotenv()
from tensorlake.sandbox import SandboxClient
def create_dummy_project(base_dir):
"""
Helper to create a simple Python project layout locally
so we have something to build/test.
"""
project_root = os.path.join(base_dir, "my_cool_project")
src_dir = os.path.join(project_root, "src", "my_cool_project")
tests_dir = os.path.join(project_root, "tests")
os.makedirs(src_dir, exist_ok=True)
os.makedirs(tests_dir, exist_ok=True)
# Create setup.py
with open(os.path.join(project_root, "setup.py"), "w") as f:
f.write("from setuptools import setup, find_packages\n"
"setup(name='my_cool_project', version='0.1.0', "
"package_dir={'': 'src'}, packages=find_packages(where='src'))")
# Create source code
with open(os.path.join(src_dir, "__init__.py"), "w") as f:
f.write("def add(a, b): return a + b")
# Create test
with open(os.path.join(tests_dir, "test_logic.py"), "w") as f:
f.write("from my_cool_project import add\ndef test_add(): assert add(2, 3) == 5")
# Create requirements.txt
with open(os.path.join(project_root, "requirements.txt"), "w") as f:
f.write("pytest\nsetuptools\nwheel\n")
return project_root
def run_ci_step(sandbox, name, command, working_dir="/workspace/project", env=None):
"""Runs a command in the sandbox, prints the output, and checks for errors."""
print(f"--- Running Step: {name} ---")
result = sandbox.run(command[0], command[1:], env=env, working_dir=working_dir)
print(f"STDOUT:\n{result.stdout}")
if result.stderr:
print(f"STDERR:\n{result.stderr}")
if result.exit_code != 0:
print(f"β Step '{name}' FAILED with exit code {result.exit_code}")
raise RuntimeError(f"CI step '{name}' failed.")
else:
print(f"β
Step '{name}' PASSED")
print("-" * (len(name) + 20))
def copy_to_sandbox(sandbox, local_path, remote_path):
"""Recursively copies a local directory to the sandbox."""
print(f"Copying {local_path} -> {remote_path} ...")
sandbox.run("mkdir", ["-p", remote_path])
for root, dirs, files in os.walk(local_path):
rel_root = os.path.relpath(root, local_path)
remote_root = remote_path if rel_root == "." else os.path.join(remote_path, rel_root)
for d in dirs:
sandbox.run("mkdir", ["-p", os.path.join(remote_root, d)])
for file in files:
local_file = os.path.join(root, file)
remote_file = os.path.join(remote_root, file)
with open(local_file, "rb") as f:
sandbox.write_file(remote_file, f.read())
async def main():
# 1. Setup local dummy project
temp_dir = tempfile.mkdtemp()
project_path = create_dummy_project(temp_dir)
print(f"Dummy project created at: {project_path}")
client = SandboxClient()
try:
# 2. Create Sandbox
with client.create_and_connect() as sandbox:
print("π Sandbox created for CI/CD pipeline.")
# 3. Upload Code
copy_to_sandbox(sandbox, project_path, "/workspace/project")
# 4. Install Dependencies
run_ci_step(
sandbox,
"Install Dependencies",
["pip", "install", "-r", "requirements.txt", "--user", "--break-system-packages"],
working_dir="/workspace/project"
)
# 5. Run Tests
run_ci_step(
sandbox,
"Run Tests",
["python", "-m", "pytest", "/workspace/project"],
env={"PYTHONPATH": "/workspace/project/src"},
)
# 6. Build Artifacts
run_ci_step(sandbox, "Build Package", ["python", "setup.py", "sdist", "bdist_wheel"])
print("\nπ CI/CD Pipeline finished successfully! π")
except Exception as e:
print(f"\nπ₯ CI/CD Pipeline FAILED: {e}")
finally:
shutil.rmtree(temp_dir)
if __name__ == "__main__":
import asyncio
asyncio.run(main())