b843c1b · Secure CI/CD Pipeline

.gitignore +10 -0

app/__init__.py +0 -0

app/app.py +45 -0

		@@ -0,0 +1,45 @@
	+	from flask import Flask, jsonify, request
	+
	+	from app.storage import TaskStore
	+
	+	app = Flask(__name__)
	+	store = TaskStore()
	+
	+
	+	@app.get("/health")
	+	def health():
	+	return jsonify(status="ok")
	+
	+
	+	@app.get("/tasks")
	+	def list_tasks():
	+	return jsonify(tasks=store.all())
	+
	+
	+	@app.post("/tasks")
	+	def create_task():
	+	payload = request.get_json(silent=True) or {}
	+	title = payload.get("title", "").strip()
	+	if not title:
	+	return jsonify(error="title is required"), 400
	+	task = store.add(title=title, done=bool(payload.get("done", False)))
	+	return jsonify(task=task), 201
	+
	+
	+	@app.get("/tasks/<int:task_id>")
	+	def get_task(task_id):
	+	task = store.get(task_id)
	+	if task is None:
	+	return jsonify(error="not found"), 404
	+	return jsonify(task=task)
	+
	+
	+	@app.delete("/tasks/<int:task_id>")
	+	def delete_task(task_id):
	+	if not store.remove(task_id):
	+	return jsonify(error="not found"), 404
	+	return "", 204
	+
	+
	+	if __name__ == "__main__":
	+	app.run(host="127.0.0.1", port=8000)

app/storage.py +28 -0

		@@ -0,0 +1,28 @@
	+	from itertools import count
	+	from threading import Lock
	+
	+
	+	class TaskStore:
	+	def __init__(self):
	+	self._tasks = {}
	+	self._ids = count(1)
	+	self._lock = Lock()
	+
	+	def all(self):
	+	with self._lock:
	+	return list(self._tasks.values())
	+
	+	def add(self, title, done=False):
	+	with self._lock:
	+	task_id = next(self._ids)
	+	task = {"id": task_id, "title": title, "done": done}
	+	self._tasks[task_id] = task
	+	return task
	+
	+	def get(self, task_id):
	+	with self._lock:
	+	return self._tasks.get(task_id)
	+
	+	def remove(self, task_id):
	+	with self._lock:
	+	return self._tasks.pop(task_id, None) is not None

pyproject.toml +23 -0

		@@ -0,0 +1,23 @@
	+	[project]
	+	name = "secure-cicd-pipeline"
	+	version = "0.1.0"
	+	requires-python = ">=3.11"
	+
	+	[tool.ruff]
	+	line-length = 100
	+	target-version = "py311"
	+
	+	[tool.ruff.lint]
	+	select = ["E", "F", "I", "B", "S"]
	+	ignore = ["S101"]
	+
	+	[tool.ruff.lint.per-file-ignores]
	+	"tests/*" = ["S"]
	+	"scripts/notify_soc.py" = ["S310"]
	+
	+	[tool.bandit]
	+	exclude_dirs = ["tests"]
	+
	+	[tool.pytest.ini_options]
	+	pythonpath = ["."]
	+	testpaths = ["tests"]

requirements.txt +1 -0

		@@ -0,0 +1 @@
	+	Flask==3.0.3

tests/__init__.py +0 -0

tests/test_app.py +44 -0

		@@ -0,0 +1,44 @@
	+	import pytest
	+
	+	from app.app import app, store
	+
	+
	+	@pytest.fixture
	+	def client():
	+	app.config["TESTING"] = True
	+	with app.test_client() as client:
	+	yield client
	+	store._tasks.clear()
	+
	+
	+	def test_health(client):
	+	resp = client.get("/health")
	+	assert resp.status_code == 200
	+	assert resp.get_json()["status"] == "ok"
	+
	+
	+	def test_create_and_fetch(client):
	+	resp = client.post("/tasks", json={"title": "review pipeline"})
	+	assert resp.status_code == 201
	+	task = resp.get_json()["task"]
	+	assert task["title"] == "review pipeline"
	+	assert task["done"] is False
	+
	+	fetched = client.get(f"/tasks/{task['id']}")
	+	assert fetched.status_code == 200
	+	assert fetched.get_json()["task"]["id"] == task["id"]
	+
	+
	+	def test_create_requires_title(client):
	+	resp = client.post("/tasks", json={})
	+	assert resp.status_code == 400
	+
	+
	+	def test_delete(client):
	+	created = client.post("/tasks", json={"title": "temp"}).get_json()["task"]
	+	assert client.delete(f"/tasks/{created['id']}").status_code == 204
	+	assert client.get(f"/tasks/{created['id']}").status_code == 404
	+
	+
	+	def test_missing_task(client):
	+	assert client.get("/tasks/999").status_code == 404

flask task api with unit tests